hzya
/
kangarooDataCenterV3
1
1
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

test(nifi): 更新测试用例以支持新的处理器创建和参数更新功能 

- 新增测试用例以验证处理器复制功能
- 新增测试用例以验证创建处理器并更新参数的功能
- 更新现有测试用例以使用新的处理器创建方法
- 优化测试用例中的参数设置和处理器配置
This commit is contained in:
liuy 2025-05-27 15:12:33 +08:00
parent 56841ca743
commit fa9def8ded
3 changed files with 157 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

70
base-buildpackage/src/test/java/com/hzya/frame/TestINifiBaseOperation.java
View File

@ -1,8 +1,12 @@
package com.hzya.frame;
import com.hzya.frame.nifiapi.model.joincreateprocessorandupdateparam.CreateProcessorAndUpdateParamJoin;
import com.hzya.frame.nifiapi.model.joingetallcontrollerservice.Component19;
import com.hzya.frame.nifiapi.model.joingetallcontrollerservice.ControllerServices19;
import com.hzya.frame.nifiapi.model.resultNeedModifyController.PortFilterResult;
import com.hzya.frame.nifiapi.model.resultprocessorsinfo.Component16;
import com.hzya.frame.nifiapi.model.resultprocessorsinfo.Config16;
import com.hzya.frame.nifiapi.operation.INifiAppOperation;
import com.hzya.frame.nifiapi.operation.INifiBaseOperation;
import org.junit.Test;
import org.junit.runner.RunWith;
@ -10,7 +14,9 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* @Author：liuyang
@ -27,6 +33,9 @@ public class TestINifiBaseOperation {
@Autowired
private INifiBaseOperation iNifiBaseOperation;
@Autowired
private INifiAppOperation iNifiAppOperation;
@Test
public void testGetNeedModifyControllerServicesByProcessGroup() throws Exception {
try {
@ -63,4 +72,65 @@ public class TestINifiBaseOperation {
e.printStackTrace();
}
}
@Test
public void testCopyProcessor() throws Exception {
try {
String targetProcessorId = "7f70897a-0196-1000-241a-3a7cdf5bf521";
String parentProcessGroupId = "76e23100-ad84-1aab-4447-b9ee936ee7d5";
String copyTargetProcessGroupId = "76e23115-ad84-1aab-7630-21c76aa3973d";
String s = iNifiBaseOperation.copyProcessor(targetProcessorId, parentProcessGroupId, copyTargetProcessGroupId);
System.out.println("新处理器id：" + s);
} catch (Exception e) {
e.printStackTrace();
}
}
@Test
public void testCreateProcessorAndUpdateParam() {
try {
Map<String, String> properties = new HashMap<>();
properties.put("File Size", "1B");
properties.put("Batch Size", "2");
properties.put("Data Format", "Binary");
properties.put("Unique FlowFiles", "true");
properties.put("character-set", "GBK");
Config16 config16 = new Config16();
config16.setProperties(properties);
Component16 component16 = new Component16();
component16.setName("修改处理器名称2");
component16.setConfig(config16);
CreateProcessorAndUpdateParamJoin createProcessorAndUpdateParamJoin = new CreateProcessorAndUpdateParamJoin();
createProcessorAndUpdateParamJoin.setComponent(component16);
createProcessorAndUpdateParamJoin.setTargetProcessorId("7f70897a-0196-1000-241a-3a7cdf5bf521");
createProcessorAndUpdateParamJoin.setCopyTargetProcessGroupId("76e23115-ad84-1aab-7630-21c76aa3973d");
createProcessorAndUpdateParamJoin.setParentProcessGroupId("76e23100-ad84-1aab-4447-b9ee936ee7d5");
String processorAndUpdateParam = iNifiAppOperation.createProcessorAndUpdateParam(createProcessorAndUpdateParamJoin);
System.out.println("新处理器id：" + processorAndUpdateParam);
} catch (Exception e) {
e.printStackTrace();
}
}
@Test
public void testgGetFindStateClearMarkProcessor() throws Exception {
try {
iNifiBaseOperation.getFindStateClearMarkProcessor("41437b87-0196-1000-6951-50e41e75c0da", "接口状态清理标记");
} catch (Exception e) {
e.printStackTrace();
}
}
@Test
public void testBatchEnabledControllerServices() throws Exception {
try {
iNifiBaseOperation.batchEnabledControllerServices("4143d803-0196-1000-968f-b06f55b43a3f", "xxxx");
} catch (Exception e) {
e.printStackTrace();
}
}
}

121
base-buildpackage/src/test/java/com/hzya/frame/TestNifiDbSync.java
View File

@ -6,6 +6,7 @@ import com.hzya.frame.nifiapi.model.joincreateconnections.Revision18;
import com.hzya.frame.nifiapi.model.joincreateconnections.SourceOrDestination18;
import com.hzya.frame.nifiapi.model.joincreateoracleapp.CreateAppInstanceJoin;
import com.hzya.frame.nifiapi.model.joincreateprocessconnection.CreateProcessorConnections;
import com.hzya.frame.nifiapi.model.joincreateprocessorandupdateparam.CreateProcessorAndUpdateParamJoin;
import com.hzya.frame.nifiapi.model.joincreparamcontext.CreateParamContextJoin;
import com.hzya.frame.nifiapi.model.joinfindneedmodifycontroller.FindNeedModifyController;
import com.hzya.frame.nifiapi.model.joingetcontroller.Component12;
@ -40,10 +41,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.*;
/**
* @Author：liuyang
@ -77,7 +75,12 @@ public class TestNifiDbSync {
String newProcessGroupsId = testCreateProcessGroups(parentProcessGroupId, processGroupsName);
//新建GenerateFlowFile处理器(开始的处理器)
String generateFlowFileId = testCreateProcessor(newProcessGroupsId);
// String generateFlowFileId = testCreateProcessor(newProcessGroupsId);
CreateProcessorAndUpdateParamJoin createProcessorAndUpdateParamJoin = new CreateProcessorAndUpdateParamJoin();
createProcessorAndUpdateParamJoin.setTargetProcessorId("02d23f3c-7003-1a1b-1cdc-e9f37874bd78");
createProcessorAndUpdateParamJoin.setParentProcessGroupId("76e23100-ad84-1aab-4447-b9ee936ee7d5");
createProcessorAndUpdateParamJoin.setCopyTargetProcessGroupId(newProcessGroupsId);
String generateFlowFileId = nifiOperation.createProcessorAndUpdateParam(createProcessorAndUpdateParamJoin);
//创建Oracle应用
//初始化实例
@ -86,11 +89,11 @@ public class TestNifiDbSync {
createOracleApp.setParentProcessGroupId("76e23100-ad84-1aab-4447-b9ee936ee7d5");
createOracleApp.setCopyTargetProcessGroupId(newProcessGroupsId);
//上下文
CreateParamContextJoin createParamContextJoin = setOracleParamContextJoin(newProcessGroupsId);
CreateParamContextJoin createParamContextJoin = setOracleParamContextJoin();
createOracleApp.setCreateParamContextJoin(createParamContextJoin);
//控制器
FindNeedModifyController findNeedModifyController = setOracleFindNeedModifyController(newProcessGroupsId);
createOracleApp.setFindNeedModifyController(findNeedModifyController);
// FindNeedModifyController findNeedModifyController = setOracleFindNeedModifyController(newProcessGroupsId);
// createOracleApp.setFindNeedModifyController(findNeedModifyController);
CreateAppInstanceResult oracleAppContext = nifiOperation.createAppContext(createOracleApp);
//创建Mysql应用
@ -103,18 +106,26 @@ public class TestNifiDbSync {
CreateParamContextJoin createParamContextJoin2 = setMysqlParamContextJoin(newProcessGroupsId);
createOracleApp2.setCreateParamContextJoin(createParamContextJoin2);
//控制器
FindNeedModifyController findNeedModifyController2 = setMysqlFindNeedModifyController(newProcessGroupsId);
createOracleApp2.setFindNeedModifyController(findNeedModifyController2);
// FindNeedModifyController findNeedModifyController2 = setMysqlFindNeedModifyController(newProcessGroupsId);
// createOracleApp2.setFindNeedModifyController(findNeedModifyController2);
CreateAppInstanceResult mysqlAppContext = nifiOperation.createAppContext(createOracleApp2);
//新建LogAttribute处理器(结束处理器)
String logAttributeId = testCreateProcessor2(newProcessGroupsId);
// String logAttributeId = testCreateProcessor2();
// String logAttributeId = null;
CreateProcessorAndUpdateParamJoin createProcessorAndUpdateParamJoin2 = new CreateProcessorAndUpdateParamJoin();
createProcessorAndUpdateParamJoin2.setTargetProcessorId("7f70897a-0196-1000-241a-3a7cdf5bf521");
createProcessorAndUpdateParamJoin2.setParentProcessGroupId("76e23100-ad84-1aab-4447-b9ee936ee7d5");
createProcessorAndUpdateParamJoin2.setCopyTargetProcessGroupId(newProcessGroupsId);
String logAttributeId = nifiOperation.createProcessorAndUpdateParam(createProcessorAndUpdateParamJoin2);
//查找对应应用的连接标记
//TYPE=INPUT_PORT
PortFilterResult oracleMark = iNifiBaseOperation.getNeedModifyControllerRelationshipByProcessGroup(oracleAppContext.getNewProcessGroupId(), null);
PortFilterResult oracleMarkInput = iNifiBaseOperation.getNeedModifyControllerRelationshipByProcessGroup(oracleAppContext.getNewProcessGroupId(), "f5228bc8-0360-41eb-a640-9f4df953937c");
PortFilterResult oracleMarkOutPut = iNifiBaseOperation.getNeedModifyControllerRelationshipByProcessGroup(oracleAppContext.getNewProcessGroupId(), "4929a9ac-291a-4ea2-a4b1-286ddf47164a");
//TYPE=OUTPUT_PORT
PortFilterResult mysqlMark = iNifiBaseOperation.getNeedModifyControllerRelationshipByProcessGroup(mysqlAppContext.getNewProcessGroupId(), null);
PortFilterResult mysqlMarkInput = iNifiBaseOperation.getNeedModifyControllerRelationshipByProcessGroup(mysqlAppContext.getNewProcessGroupId(), "be05a6ad-a440-43ed-b4c0-d1edcf03334f");
PortFilterResult mysqlMarkOutput = iNifiBaseOperation.getNeedModifyControllerRelationshipByProcessGroup(mysqlAppContext.getNewProcessGroupId(), "7ac85368-f2c9-48df-ab5b-ea610449083d");
//批量创建连接关系
//连接关系一:开始处理器->Oracle应用
@ -124,7 +135,7 @@ public class TestNifiDbSync {
source.setType("PROCESSOR");
SourceOrDestination18 destination = new SourceOrDestination18();
destination.setId(oracleMark.getInputPorts().get(0).getId());
destination.setId(oracleMarkInput.getInputPorts().get(0).getId());
destination.setGroupId(oracleAppContext.getNewProcessGroupId());
destination.setType("INPUT_PORT");
@ -148,12 +159,12 @@ public class TestNifiDbSync {
//连接关系二：Oracle应用->Mysql应用
SourceOrDestination18 source2 = new SourceOrDestination18();
source2.setId(oracleMark.getOutputPorts().get(0).getId());
source2.setId(oracleMarkOutPut.getOutputPorts().get(0).getId());
source2.setGroupId(oracleAppContext.getNewProcessGroupId());
source2.setType("OUTPUT_PORT");
SourceOrDestination18 destination2 = new SourceOrDestination18();
destination2.setId(mysqlMark.getInputPorts().get(0).getId());
destination2.setId(mysqlMarkInput.getInputPorts().get(0).getId());
destination2.setGroupId(mysqlAppContext.getNewProcessGroupId());
destination2.setType("INPUT_PORT");
@ -175,7 +186,7 @@ public class TestNifiDbSync {
//连接关系三：Mysql应用->结束处理器
SourceOrDestination18 source3 = new SourceOrDestination18();
source3.setId(mysqlMark.getOutputPorts().get(0).getId());
source3.setId(mysqlMarkOutput.getOutputPorts().get(0).getId());
source3.setGroupId(mysqlAppContext.getNewProcessGroupId());
source3.setType("OUTPUT_PORT");
@ -205,6 +216,12 @@ public class TestNifiDbSync {
createProcessorConnections.setCreateConnections18(createConnections18List);
iNifiBaseOperation.batchCreateProcessorConnections(createProcessorConnections);
//去掉流程组内对应的处理器旧状态
// List<String> stringList = new ArrayList<>();
// stringList.add(oracleAppContext.getNewProcessGroupId());
// stringList.add(mysqlAppContext.getNewProcessGroupId());
// iNifiBaseOperation.clearProcessGroupState(stringList);
//启动业务处理器
ProcessGroupsInfoResult9 processGroups = nifiApiService.getProcessGroups(newProcessGroupsId);
Revision9 revision = processGroups.getRevision();
@ -221,7 +238,7 @@ public class TestNifiDbSync {
}
public CreateParamContextJoin setOracleParamContextJoin(String processGroupsId) throws Exception {
public CreateParamContextJoin setOracleParamContextJoin() throws Exception {
Revision revision = new Revision();
revision.setVersion(0);
@ -229,31 +246,31 @@ public class TestNifiDbSync {
parameter.setSensitive(false);
parameter.setName("additionalWhereClause");
parameter.setValue("1=1");
parameter.setDescription("参数描述");
parameter.setDescription("where条件");
Parameter2 parameter2 = new Parameter2();
parameter2.setSensitive(false);
parameter2.setName("columnsToReturn");
parameter2.setValue("CSALEID,VRECEIPTCODE,NHEADSUMMNY,TS");
parameter2.setDescription("参数描述");
parameter2.setValue("pk_invbasdoc,invcode,ts");
parameter2.setDescription("需要查询的表字段");
Parameter2 parameter3 = new Parameter2();
parameter3.setSensitive(false);
parameter3.setName("customOrderByColumn");
parameter3.setValue("ts asc");
parameter3.setDescription("参数描述");
parameter3.setDescription("排序规则");
Parameter2 parameter4 = new Parameter2();
parameter4.setSensitive(false);
parameter4.setName("maximumValueColumns");
parameter4.setValue("ts");
parameter4.setDescription("参数描述");
parameter4.setDescription("分片字段");
Parameter2 parameter5 = new Parameter2();
parameter5.setSensitive(false);
parameter5.setName("sourceTbName");
parameter5.setValue("U8C241231.SO_SALE");
parameter5.setDescription("参数描述");
parameter5.setValue("U8C241231.BD_INVBASDOC");
parameter5.setDescription("查询表名称");
Parameters2 parameters = new Parameters2();
parameters.setParameter(parameter);
@ -278,7 +295,7 @@ public class TestNifiDbSync {
parametersList.add(parameters5);
Component3 component = new Component3();
component.setName("oracle参数环境2505221548");
component.setName("oracle参数环境2505241627");
component.setParameters(parametersList);
ParameterContextsJoin parameterContextsJoin = new ParameterContextsJoin();
@ -287,7 +304,7 @@ public class TestNifiDbSync {
CreateParamContextJoin createParamContextJoin = new CreateParamContextJoin();
createParamContextJoin.setParameterContextsJoin(parameterContextsJoin);
createParamContextJoin.setProcessGroupsId(processGroupsId);
// createParamContextJoin.setProcessGroupsId(processGroupsId);
return createParamContextJoin;
}
@ -295,23 +312,29 @@ public class TestNifiDbSync {
Revision revision = new Revision();
revision.setVersion(0);
Parameter2 parameter = new Parameter2();
parameter.setSensitive(false);
parameter.setName("Release Signal Identifier");
parameter.setValue("table_creation");
parameter.setDescription("参数描述");
// Parameter2 parameter = new Parameter2();
// parameter.setSensitive(false);
// parameter.setName("Release Signal Identifier");
// parameter.setValue("table_creation");
// parameter.setDescription("参数描述");
Parameter2 parameter2 = new Parameter2();
parameter2.setSensitive(false);
parameter2.setName("autoTableCreation");
parameter2.setValue("Y");
parameter2.setDescription("参数描述");
parameter2.setDescription("是否自动建表");
Parameter2 parameter3 = new Parameter2();
parameter3.setSensitive(false);
parameter3.setName("mappingRelationship");
parameter3.setValue("[\n" + " {\n" + " \"targetTbName\": \"iep_so_sale\",\n" + " \"targetDBType\": \"mysql8.0.44\",\n" + " \"writeType\": \"1\",\n" + " \"fieldRelationship\": [\n" + " {\n" + " \"sourceFieldName\": \"CSALEID\",\n" + " \"sourceFieldType\": \"CHAR(20)\",\n" + " \"targetFieldName\": \"csaleid\",\n" + " \"targetFieldType\": \"varchar(20)\",\n" + " \"targetFieldNameNotes\": \"销售主表id\",\n" + " \"primaryKey\": \"Y\"\n" + " },\n" + " {\n" + " \"sourceFieldName\": \"VRECEIPTCODE\",\n" + " \"sourceFieldType\": \"VARCHAR(30)\",\n" + " \"targetFieldName\": \"vreceiptcode\",\n" + " \"targetFieldType\": \"varchar(20)\",\n" + " \"targetFieldNameNotes\": \"单据号\"\n" + " },\n" + " {\n" + " \"sourceFieldName\": \"NHEADSUMMNY\",\n" + " \"sourceFieldType\": \"DECIMAL(13,8)\",\n" + " \"targetFieldName\": \"nheadsummny\",\n" + " \"targetFieldType\": \"decimal(20,10)\",\n" + " \"targetFieldNameNotes\": \"金额合计\"\n" + " },\n" + " {\n" + " \"sourceFieldName\": \"TS\",\n" + " \"sourceFieldType\": \"CHAR(19)\",\n" + " \"targetFieldName\": \"ts\",\n" + " \"targetFieldType\": \"varchar(20)\",\n" + " \"targetFieldNameNotes\": \"时间戳\"\n" + " }\n" + " ]\n" + " }\n" + " ]");
parameter3.setDescription("参数描述");
parameter3.setValue("[\n" + " {\n" + " \"targetTbName\": \"iep_bd_invbasdoc\",\n" + " \"targetDBType\": \"mysql8.0.44\",\n" + " \"writeType\": \"1\",\n" + " \"fieldRelationship\": [\n" + " {\n" + " \"sourceFieldName\": \"PK_INVBASDOC\",\n" + " \"sourceFieldType\": \"CHAR(20)\",\n" + " \"targetFieldName\": \"pk_invbasdoc\",\n" + " \"targetFieldType\": \"varchar(20)\",\n" + " \"targetFieldNameNotes\": \"存货档案id\",\n" + " \"primaryKey\": \"Y\"\n" + " },\n" + " {\n" + " \"sourceFieldName\": \"INVCODE\",\n" + " \"sourceFieldType\": \"VARCHAR(30)\",\n" + " \"targetFieldName\": \"invcode\",\n" + " \"targetFieldType\": \"varchar(20)\",\n" + " \"targetFieldNameNotes\": \"单据号\"\n" + " },\n" + " {\n" + " \"sourceFieldName\": \"TS\",\n" + " \"sourceFieldType\": \"CHAR(19)\",\n" + " \"targetFieldName\": \"ts\",\n" + " \"targetFieldType\": \"varchar(20)\",\n" + " \"targetFieldNameNotes\": \"时间戳\"\n" + " }\n" + " ]\n" + " }\n" + "]");
parameter3.setDescription("字段对照关系");
Parameter2 parameter4 = new Parameter2();
parameter4.setSensitive(false);
parameter4.setName("releaseSignalIdentifier");
parameter4.setValue(UUID.randomUUID().toString());
parameter4.setDescription("建表信号，传随机值！");
// Parameter2 parameter4 = new Parameter2();
// parameter4.setSensitive(false);
@ -326,7 +349,7 @@ public class TestNifiDbSync {
// parameter5.setDescription("参数描述");
Parameters2 parameters = new Parameters2();
parameters.setParameter(parameter);
parameters.setParameter(parameter4);
Parameters2 parameters2 = new Parameters2();
parameters2.setParameter(parameter2);
@ -348,7 +371,7 @@ public class TestNifiDbSync {
// parametersList.add(parameters5);
Component3 component = new Component3();
component.setName("mysql参数环境2505230915");
component.setName("mysql参数环境2505241627");
component.setParameters(parametersList);
ParameterContextsJoin parameterContextsJoin = new ParameterContextsJoin();
@ -478,7 +501,8 @@ public class TestNifiDbSync {
String mysqlProcessGroupId = testCreateSnippetInstance(processGroupsId, parentProcessGroupId, mysqlSnippetId);
//新建LogAttribute处理器(结束处理器)
String logAttributeId = testCreateProcessor2(processGroupsId);
// String logAttributeId = testCreateProcessor2();
String logAttributeId = null;
//修改Oracle应用数据源
//修改Oracle应用上下文环境
@ -596,8 +620,9 @@ public class TestNifiDbSync {
/**
* 创建LogAttribute处理器
*/
public String testCreateProcessor2(String groupId) throws Exception {
String processorId = null;
@Test
public void testCreateProcessor2() throws Exception {
// String processorId = null;
try {
Revision15 revision15 = new Revision15();
revision15.setVersion("0");
@ -625,18 +650,28 @@ public class TestNifiDbSync {
component15.setConfig(config15);
// component15.setPosition(position15);
//定义连接关系
Relationships15 relationships15 = new Relationships15();
relationships15.setName("success");
relationships15.setAutoTerminate(true);
// relationships15.setRetry(true);
// relationships15.setDescription("123");
List<Relationships15> relationships15List = new ArrayList<>();
relationships15List.add(relationships15);
component15.setRelationships(relationships15List);
CreateProcessorJoin15 createProcessorJoin15 = new CreateProcessorJoin15();
createProcessorJoin15.setComponent(component15);
createProcessorJoin15.setRevision(revision15);
// String groupId = "019610d1-f933-1f70-16ea-3cf005012c4b";
String groupId = "76e23115-ad84-1aab-7630-21c76aa3973d";
CreateProcess16 processor = nifiApiService.createProcessor(groupId, createProcessorJoin15);
// System.out.println("处理器id：" + processor.getId());
processorId = processor.getId();
// processorId = processor.getId();
} catch (Exception e) {
e.printStackTrace();
}
return processorId;
// return processorId;
}
/**

10
base-buildpackage/src/test/java/com/hzya/frame/TestNifiServiceApi.java
View File

@ -896,7 +896,7 @@ public class TestNifiServiceApi {
public void testGetProcessor() throws Exception {
try {
// String processId = "01961137-f933-1f70-4f08-6fc7041f99bd";
String processId2 = "7f70897a-0196-1000-241a-3a7cdf5bf521";
String processId2 = "02d23f3c-7003-1a1b-1cdc-e9f37874bd78";
ProcessorsInfo16 processor = nifiApiService.getProcessor(processId2);
String id = processor.getId();
System.out.println(id);
@ -1087,4 +1087,12 @@ public class TestNifiServiceApi {
// e.printStackTrace();
// }
// }
@Test
public void testClearProcessorRequests() throws Exception {
try {
nifiApiService.clearProcessorRequests("019617a7-f933-1f70-9747-dac20033270f");
} catch (Exception e) {
e.printStackTrace();
}
}
}