github
/
DolphinScheduler
mirror of https://gitee.com/dolphinscheduler/DolphinScheduler.git
1
0
Fork 0
Code Issues Releases Wiki Activity
Browse Source

[python] Fix submit and run error (#10792) 

because we add permission, so it failed out
createQueue method. this patch fix it and
do some refactor of our tenant and queue
validator code

(cherry picked from commit ae6aa53f96)
3.0.0/version-upgrade
Jiajie Zhong 2 years ago
parent
cec5e88319
commit
d3f28c8411
13 changed files with 487 additions and 415 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 2
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java
  2. 6
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/exceptions/ApiExceptionHandler.java
  3. 7
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/exceptions/ServiceException.java
  4. 37
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java
  5. 10
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/QueueService.java
  6. 18
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TenantService.java
  7. 187
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/QueueServiceImpl.java
  8. 203
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TenantServiceImpl.java
  9. 58
      dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/QueueControllerTest.java
  10. 144
      dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/QueueServiceTest.java
  11. 169
      dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TenantServiceTest.java
  12. 7
      dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/QueueMapper.java
  13. 6
      dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/QueueMapper.xml

2
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java
Unescape View File

@ -194,7 +194,7 @@ public enum Status {
QUERY_WORKFLOW_LINEAGE_ERROR(10161, "query workflow lineage error", "查询血缘失败"),
QUERY_AUTHORIZED_AND_USER_CREATED_PROJECT_ERROR(10162, "query authorized and user created project error error", "查询授权的和用户创建的项目错误"),
DELETE_PROCESS_DEFINITION_BY_CODE_FAIL(10163, "delete process definition by code fail, for there are {0} process instances in executing using it", "删除工作流定义失败，有[{0}]个运行中的工作流实例正在使用"),
CHECK_OS_TENANT_CODE_ERROR(10164, "Please enter the English os tenant code", "请输入英文操作系统租户"),
CHECK_OS_TENANT_CODE_ERROR(10164, "Tenant code invalid, should follow linux's users naming conventions", "非法的租户名，需要遵守 Linux 用户命名规范"),
FORCE_TASK_SUCCESS_ERROR(10165, "force task success error", "强制成功任务实例错误"),
TASK_INSTANCE_STATE_OPERATION_ERROR(10166, "the status of task instance {0} is {1},Cannot perform force success operation", "任务实例[{0}]的状态是[{1}]，无法执行强制成功操作"),
DATASOURCE_TYPE_NOT_EXIST(10167, "data source type not exist", "数据源类型不存在"),

6
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/exceptions/ApiExceptionHandler.java
Unescape View File

@ -36,6 +36,12 @@ public class ApiExceptionHandler {
private static final Logger logger = LoggerFactory.getLogger(ApiExceptionHandler.class);
@ExceptionHandler(ServiceException.class)
public Result exceptionHandler(ServiceException e, HandlerMethod hm) {
logger.error("ServiceException: ", e);
return new Result(e.getCode(), e.getMessage());
}
@ExceptionHandler(Exception.class)
public Result exceptionHandler(Exception e, HandlerMethod hm) {
ApiException ce = hm.getMethodAnnotation(ApiException.class);

7
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/exceptions/ServiceException.java
Unescape View File

@ -18,6 +18,8 @@ package org.apache.dolphinscheduler.api.exceptions;
import org.apache.dolphinscheduler.api.enums.Status;
import java.text.MessageFormat;
/**
* service exception
@ -37,6 +39,11 @@ public class ServiceException extends RuntimeException {
this.code = status.getCode();
}
public ServiceException(Status status, Object... formatter) {
super(MessageFormat.format(status.getMsg(), formatter));
this.code = status.getCode();
}
public ServiceException(Integer code,String message) {
super(message);
this.code = code;

37
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java
Unescape View File

@ -22,7 +22,6 @@ import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.service.ExecutorService;
import org.apache.dolphinscheduler.api.service.ProcessDefinitionService;
import org.apache.dolphinscheduler.api.service.ProjectService;
import org.apache.dolphinscheduler.api.service.QueueService;
import org.apache.dolphinscheduler.api.service.ResourcesService;
import org.apache.dolphinscheduler.api.service.SchedulerService;
import org.apache.dolphinscheduler.api.service.TaskDefinitionService;
@ -114,9 +113,6 @@ public class PythonGateway {
@Autowired
private UsersService usersService;
@Autowired
private QueueService queueService;
@Autowired
private ResourcesService resourceService;
@ -395,37 +391,8 @@ public class PythonGateway {
}
}
public Map<String, Object> createQueue(String name, String queueName) {
Result<Object> verifyQueueExists = queueService.verifyQueue(name, queueName);
if (verifyQueueExists.getCode() == 0) {
return queueService.createQueue(dummyAdminUser, name, queueName);
} else {
Map<String, Object> result = new HashMap<>();
// TODO function putMsg do not work here
result.put(Constants.STATUS, Status.SUCCESS);
result.put(Constants.MSG, Status.SUCCESS.getMsg());
return result;
}
}
public Map<String, Object> createTenant(String tenantCode, String desc, String queueName) throws Exception {
if (tenantService.checkTenantExists(tenantCode)) {
Map<String, Object> result = new HashMap<>();
// TODO function putMsg do not work here
result.put(Constants.STATUS, Status.SUCCESS);
result.put(Constants.MSG, Status.SUCCESS.getMsg());
return result;
} else {
Result<Object> verifyQueueExists = queueService.verifyQueue(queueName, queueName);
if (verifyQueueExists.getCode() == 0) {
// TODO why create do not return id?
queueService.createQueue(dummyAdminUser, queueName, queueName);
}
Map<String, Object> result = queueService.queryQueueName(queueName);
List<Queue> queueList = (List<Queue>) result.get(Constants.DATA_LIST);
Queue queue = queueList.get(0);
return tenantService.createTenant(dummyAdminUser, tenantCode, queue.getId(), desc);
}
public Tenant createTenant(String tenantCode, String desc, String queueName) {
return tenantService.createTenantIfNotExists(tenantCode, desc, queueName, queueName);
}
public void createUser(String userName,

10
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/QueueService.java
Unescape View File

@ -18,6 +18,7 @@
package org.apache.dolphinscheduler.api.service;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.dao.entity.Queue;
import org.apache.dolphinscheduler.dao.entity.User;
import java.util.Map;
@ -77,11 +78,14 @@ public interface QueueService {
Result<Object> verifyQueue(String queue, String queueName);
/**
* query queue by queueName
* Make sure queue with given name exists, and create the queue if not exists
*
* ONLY for python gateway server, and should not use this in web ui function
*
* @param queue queue value
* @param queueName queue name
* @return queue object for provide queue name
* @return Queue object
*/
Map<String, Object> queryQueueName(String queueName);
Queue createQueueIfNotExists(String queue, String queueName);
}

18
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TenantService.java
Unescape View File

@ -18,6 +18,7 @@
package org.apache.dolphinscheduler.api.service;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.dao.entity.Tenant;
import org.apache.dolphinscheduler.dao.entity.User;
import java.util.Map;
@ -94,18 +95,23 @@ public interface TenantService {
Result verifyTenantCode(String tenantCode);
/**
* check if provide tenant code object exists
* query tenant by tenant code
*
* @param tenantCode tenant code
* @return true if tenant code exists, false if not
* @return tenant list
*/
boolean checkTenantExists(String tenantCode);
Map<String, Object> queryByTenantCode(String tenantCode);
/**
* query tenant by tenant code
* Make sure tenant with given name exists, and create the tenant if not exists
*
* ONLY for python gateway server, and should not use this in web ui function
*
* @param tenantCode tenant code
* @return tenant list
* @param desc The description of tenant object
* @param queue The value of queue which current tenant belong
* @param queueName The name of queue which current tenant belong
* @return Tenant object
*/
Map<String, Object> queryByTenantCode(String tenantCode);
Tenant createTenantIfNotExists(String tenantCode, String desc, String queue, String queueName);
}

187
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/QueueServiceImpl.java
Unescape View File

@ -17,30 +17,40 @@
package org.apache.dolphinscheduler.api.service.impl;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import org.apache.commons.lang3.StringUtils;
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.exceptions.ServiceException;
import org.apache.dolphinscheduler.api.service.QueueService;
import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.AuthorizationType;
import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.dao.entity.Queue;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.QueueMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.commons.lang.StringUtils;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.YARN_QUEUE_CREATE;
import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.YARN_QUEUE_UPDATE;
/**
* queue service impl
@ -56,6 +66,44 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
@Autowired
private UserMapper userMapper;
/**
* Valid both queue and queueName when we want to create or update queue object
*
* @param queue queue value
* @param queueName queue name
*/
private void queueValid(String queue, String queueName) throws ServiceException {
if (StringUtils.isEmpty(queue)) {
throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE);
} else if (StringUtils.isEmpty(queueName)) {
throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE_NAME);
} else if (checkQueueExist(queue)) {
throw new ServiceException(Status.QUEUE_VALUE_EXIST, queue);
} else if (checkQueueNameExist(queueName)) {
throw new ServiceException(Status.QUEUE_NAME_EXIST, queueName);
}
}
/**
* Insert one single new Queue record to database
*
* @param queue queue value
* @param queueName queue name
* @return Queue
*/
private Queue createObjToDB(String queue, String queueName) {
Queue queueObj = new Queue();
Date now = new Date();
queueObj.setQueue(queue);
queueObj.setQueueName(queueName);
queueObj.setCreateTime(now);
queueObj.setUpdateTime(now);
// save
queueMapper.insert(queueObj);
return queueObj;
}
/**
* query queue list
*
@ -121,39 +169,11 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
if (isNotAdmin(loginUser, result)) {
return result;
}
queueValid(queue, queueName);
if (StringUtils.isEmpty(queue)) {
putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE);
return result;
}
if (StringUtils.isEmpty(queueName)) {
putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE_NAME);
return result;
}
if (checkQueueNameExist(queueName)) {
putMsg(result, Status.QUEUE_NAME_EXIST, queueName);
return result;
}
if (checkQueueExist(queue)) {
putMsg(result, Status.QUEUE_VALUE_EXIST, queue);
return result;
}
Queue queueObj = new Queue();
Date now = new Date();
queueObj.setQueue(queue);
queueObj.setQueueName(queueName);
queueObj.setCreateTime(now);
queueObj.setUpdateTime(now);
queueMapper.insert(queueObj);
result.put(Constants.DATA_LIST, queueObj);
Queue newQueue = createObjToDB(queue, queueName);
result.put(Constants.DATA_LIST, newQueue);
putMsg(result, Status.SUCCESS);
return result;
}
@ -173,39 +193,16 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
return result;
}
if (StringUtils.isEmpty(queue)) {
putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE);
return result;
}
if (StringUtils.isEmpty(queueName)) {
putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE_NAME);
return result;
}
queueValid(queue, queueName);
Queue queueObj = queueMapper.selectById(id);
if (queueObj == null) {
putMsg(result, Status.QUEUE_NOT_EXIST, id);
return result;
if (Objects.isNull(queueObj)) {
throw new ServiceException(Status.QUEUE_NOT_EXIST, queue);
}
// whether queue value or queueName is changed
if (queue.equals(queueObj.getQueue()) && queueName.equals(queueObj.getQueueName())) {
putMsg(result, Status.NEED_NOT_UPDATE_QUEUE);
return result;
}
// check queue name is exist
if (!queueName.equals(queueObj.getQueueName())
&& checkQueueNameExist(queueName)) {
putMsg(result, Status.QUEUE_NAME_EXIST, queueName);
return result;
}
// check queue value is exist
if (!queue.equals(queueObj.getQueue()) && checkQueueExist(queue)) {
putMsg(result, Status.QUEUE_VALUE_EXIST, queue);
return result;
throw new ServiceException(Status.NEED_NOT_UPDATE_QUEUE);
}
// check old queue using by any user
@ -238,53 +235,8 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
@Override
public Result<Object> verifyQueue(String queue, String queueName) {
Result<Object> result = new Result<>();
queueValid(queue, queueName);
if (StringUtils.isEmpty(queue)) {
putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE);
return result;
}
if (StringUtils.isEmpty(queueName)) {
putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE_NAME);
return result;
}
if (checkQueueNameExist(queueName)) {
putMsg(result, Status.QUEUE_NAME_EXIST, queueName);
return result;
}
if (checkQueueExist(queue)) {
putMsg(result, Status.QUEUE_VALUE_EXIST, queue);
return result;
}
putMsg(result, Status.SUCCESS);
return result;
}
/**
* query queue by queueName
*
* @param queueName queue name
* @return queue object for provide queue name
*/
@Override
public Map<String, Object> queryQueueName(String queueName) {
Map<String, Object> result = new HashMap<>();
if (StringUtils.isEmpty(queueName)) {
putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE_NAME);
return result;
}
if (!checkQueueNameExist(queueName)) {
putMsg(result, Status.QUEUE_NOT_EXIST, queueName);
return result;
}
List<Queue> queueList = queueMapper.queryQueueName(queueName);
result.put(Constants.DATA_LIST, queueList);
putMsg(result, Status.SUCCESS);
return result;
}
@ -324,4 +276,23 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
return !oldQueue.equals(newQueue) && userMapper.existUser(oldQueue) == Boolean.TRUE;
}
/**
* Make sure queue with given name exists, and create the queue if not exists
*
* ONLY for python gateway server, and should not use this in web ui function
*
* @param queue queue value
* @param queueName queue name
* @return Queue object
*/
@Override
public Queue createQueueIfNotExists(String queue, String queueName) {
queueValid(queue, queueName);
Queue existsQueue = queueMapper.queryQueueName(queue, queueName);
if (Objects.isNull(existsQueue)) {
return createObjToDB(queue, queueName);
}
return existsQueue;
}
}

203
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TenantServiceImpl.java
Unescape View File

@ -20,31 +20,43 @@ package org.apache.dolphinscheduler.api.service.impl;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.exceptions.ServiceException;
import org.apache.dolphinscheduler.api.service.QueueService;
import org.apache.dolphinscheduler.api.service.TenantService;
import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.RegexUtils;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.AuthorizationType;
import org.apache.dolphinscheduler.common.storage.StorageOperate;
import org.apache.dolphinscheduler.common.utils.PropertyUtils;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.Queue;
import org.apache.dolphinscheduler.dao.entity.Tenant;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.*;
import static org.apache.dolphinscheduler.common.Constants.TENANT_FULL_NAME_MAX_LENGTH;
/**
@ -53,6 +65,8 @@ import static org.apache.dolphinscheduler.common.Constants.TENANT_FULL_NAME_MAX_
@Service
public class TenantServiceImpl extends BaseServiceImpl implements TenantService {
private static final Logger logger = LoggerFactory.getLogger(TenantServiceImpl.class);
@Autowired
private TenantMapper tenantMapper;
@ -65,9 +79,53 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
@Autowired
private UserMapper userMapper;
@Autowired
private QueueService queueService;
@Autowired(required = false)
private StorageOperate storageOperate;
/**
* Valid tenantCode when we want to create or update tenant object
*
* @param tenantCode Tenant code of tenant object
* @return Optional of Status map
*/
private void tenantCodeValid(String tenantCode) throws ServiceException {
Map<String, Object> result = new HashMap<>();
if (StringUtils.isEmpty(tenantCode)) {
throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, tenantCode);
} else if (StringUtils.length(tenantCode) > TENANT_FULL_NAME_MAX_LENGTH) {
throw new ServiceException(Status.TENANT_FULL_NAME_TOO_LONG_ERROR);
} else if (!RegexUtils.isValidLinuxUserName(tenantCode)) {
throw new ServiceException(Status.CHECK_OS_TENANT_CODE_ERROR);
} else if (checkTenantExists(tenantCode)) {
throw new ServiceException(Status.OS_TENANT_CODE_EXIST, tenantCode);
}
}
/**
* Insert one single new Tenant record to database
*
* @param tenantCode new Tenant object tenant code
* @param desc new Tenant object description
* @param queueId The Queue id of new Tenant object
* @return Tenant
*/
private Tenant createObjToDB(String tenantCode, String desc, int queueId) {
Tenant tenant = new Tenant();
Date now = new Date();
tenant.setTenantCode(tenantCode);
tenant.setQueueId(queueId);
tenant.setDescription(desc);
tenant.setCreateTime(now);
tenant.setUpdateTime(now);
// save
tenantMapper.insert(tenant);
return tenant;
}
/**
* create tenant
*
@ -79,51 +137,27 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
* @throws Exception exception
*/
@Override
@Transactional(rollbackFor = Exception.class)
@Transactional
public Map<String, Object> createTenant(User loginUser,
String tenantCode,
int queueId,
String desc) throws Exception {
Map<String, Object> result = new HashMap<>();
result.put(Constants.STATUS, false);
if (isNotAdmin(loginUser, result)) {
return result;
}
if(StringUtils.length(tenantCode) > TENANT_FULL_NAME_MAX_LENGTH){
putMsg(result, Status.TENANT_FULL_NAME_TOO_LONG_ERROR);
return result;
if (!canOperatorPermissions(loginUser,null, AuthorizationType.TENANT, TENANT_CREATE)) {
throw new ServiceException(Status.USER_NO_OPERATION_PERM);
}
if (!RegexUtils.isValidLinuxUserName(tenantCode)) {
putMsg(result, Status.CHECK_OS_TENANT_CODE_ERROR);
return result;
}
if (checkTenantExists(tenantCode)) {
putMsg(result, Status.OS_TENANT_CODE_EXIST, tenantCode);
return result;
}
Tenant tenant = new Tenant();
Date now = new Date();
tenant.setTenantCode(tenantCode);
tenant.setQueueId(queueId);
tenant.setDescription(desc);
tenant.setCreateTime(now);
tenant.setUpdateTime(now);
// save
tenantMapper.insert(tenant);
tenantCodeValid(tenantCode);
Tenant newTenant = createObjToDB(tenantCode, desc, queueId);
// if storage startup
if (PropertyUtils.getResUploadStartupState()) {
storageOperate.createTenantDirIfNotExists(tenantCode);
}
result.put(Constants.DATA_LIST, tenant);
permissionPostHandle(AuthorizationType.TENANT, loginUser.getId(), Collections.singletonList(newTenant.getId()), logger);
result.put(Constants.DATA_LIST, newTenant);
putMsg(result, Status.SUCCESS);
return result;
}
@ -140,16 +174,18 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
public Result<Object> queryTenantList(User loginUser, String searchVal, Integer pageNo, Integer pageSize) {
Result<Object> result = new Result<>();
if (!isAdmin(loginUser)) {
putMsg(result, Status.USER_NO_OPERATION_PERM);
PageInfo<Tenant> pageInfo = new PageInfo<>(pageNo, pageSize);
Set<Integer> ids = resourcePermissionCheckService.userOwnedResourceIdsAcquisition(AuthorizationType.TENANT, loginUser.getId(), logger);
if (ids.isEmpty()) {
result.setData(pageInfo);
putMsg(result, Status.SUCCESS);
return result;
}
Page<Tenant> page = new Page<>(pageNo, pageSize);
IPage<Tenant> tenantIPage = tenantMapper.queryTenantPaging(page, searchVal);
PageInfo<Tenant> pageInfo = new PageInfo<>(pageNo, pageSize);
pageInfo.setTotal((int) tenantIPage.getTotal());
pageInfo.setTotalList(tenantIPage.getRecords());
IPage<Tenant> tenantPage = tenantMapper.queryTenantPaging(page, new ArrayList<>(ids), searchVal);
pageInfo.setTotal((int) tenantPage.getTotal());
pageInfo.setTotalList(tenantPage.getRecords());
result.setData(pageInfo);
putMsg(result, Status.SUCCESS);
return result;
@ -171,41 +207,29 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
String desc) throws Exception {
Map<String, Object> result = new HashMap<>();
result.put(Constants.STATUS, false);
if (isNotAdmin(loginUser, result)) {
return result;
if (!canOperatorPermissions(loginUser,null, AuthorizationType.TENANT,TENANT_UPDATE)) {
throw new ServiceException(Status.USER_NO_OPERATION_PERM);
}
Tenant tenant = tenantMapper.queryById(id);
if (tenant == null) {
putMsg(result, Status.TENANT_NOT_EXIST);
return result;
if (Objects.isNull(tenant)) {
throw new ServiceException(Status.TENANT_NOT_EXIST);
}
tenantCodeValid(tenantCode);
// updateProcessInstance tenant
/**
* if the tenant code is modified, the original resource needs to be copied to the new tenant.
*/
if (!tenant.getTenantCode().equals(tenantCode)) {
if (checkTenantExists(tenantCode)) {
// if hdfs startup
if (PropertyUtils.getResUploadStartupState()) {
if (!tenant.getTenantCode().equals(tenantCode) && PropertyUtils.getResUploadStartupState()) {
storageOperate.createTenantDirIfNotExists(tenantCode);
}
} else {
putMsg(result, Status.OS_TENANT_CODE_HAS_ALREADY_EXISTS);
return result;
}
}
Date now = new Date();
if (!StringUtils.isEmpty(tenantCode)) {
tenant.setTenantCode(tenantCode);
}
if (queueId != 0) {
tenant.setQueueId(queueId);
}
@ -213,8 +237,7 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
tenant.setUpdateTime(now);
tenantMapper.updateById(tenant);
result.put(Constants.STATUS, Status.SUCCESS);
result.put(Constants.MSG, Status.SUCCESS.getMsg());
putMsg(result, Status.SUCCESS);
return result;
}
@ -227,37 +250,33 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
* @throws Exception exception
*/
@Override
@Transactional(rollbackFor = Exception.class)
@Transactional
public Map<String, Object> deleteTenantById(User loginUser, int id) throws Exception {
Map<String, Object> result = new HashMap<>();
if (isNotAdmin(loginUser, result)) {
return result;
if (!canOperatorPermissions(loginUser,null, AuthorizationType.TENANT,TENANT_DELETE)) {
throw new ServiceException(Status.USER_NO_OPERATION_PERM);
}
Tenant tenant = tenantMapper.queryById(id);
if (tenant == null) {
putMsg(result, Status.TENANT_NOT_EXIST);
return result;
if (Objects.isNull(tenant)) {
throw new ServiceException(Status.TENANT_NOT_EXIST);
}
List<ProcessInstance> processInstances = getProcessInstancesByTenant(tenant);
if (CollectionUtils.isNotEmpty(processInstances)) {
putMsg(result, Status.DELETE_TENANT_BY_ID_FAIL, processInstances.size());
return result;
throw new ServiceException(Status.DELETE_TENANT_BY_ID_FAIL, processInstances.size());
}
List<ProcessDefinition> processDefinitions =
processDefinitionMapper.queryDefinitionListByTenant(tenant.getId());
if (CollectionUtils.isNotEmpty(processDefinitions)) {
putMsg(result, Status.DELETE_TENANT_BY_ID_FAIL_DEFINES, processDefinitions.size());
return result;
throw new ServiceException(Status.DELETE_TENANT_BY_ID_FAIL_DEFINES, processDefinitions.size());
}
List<User> userList = userMapper.queryUserListByTenant(tenant.getId());
if (CollectionUtils.isNotEmpty(userList)) {
putMsg(result, Status.DELETE_TENANT_BY_ID_FAIL_USERS, userList.size());
return result;
throw new ServiceException(Status.DELETE_TENANT_BY_ID_FAIL_USERS, userList.size());
}
// if resource upload startup
@ -286,11 +305,15 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
public Map<String, Object> queryTenantList(User loginUser) {
Map<String, Object> result = new HashMap<>();
List<Tenant> resourceList = tenantMapper.selectList(null);
Set<Integer> ids = resourcePermissionCheckService.userOwnedResourceIdsAcquisition(AuthorizationType.TENANT, loginUser.getId(), logger);
if (ids.isEmpty()) {
result.put(Constants.DATA_LIST, Collections.emptyList());
putMsg(result, Status.SUCCESS);
return result;
}
List<Tenant> resourceList = tenantMapper.selectBatchIds(ids);
result.put(Constants.DATA_LIST, resourceList);
putMsg(result, Status.SUCCESS);
return result;
}
@ -304,10 +327,9 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
public Result<Object> verifyTenantCode(String tenantCode) {
Result<Object> result = new Result<>();
if (checkTenantExists(tenantCode)) {
putMsg(result, Status.OS_TENANT_CODE_EXIST, tenantCode);
} else {
putMsg(result, Status.SUCCESS);
throw new ServiceException(Status.OS_TENANT_CODE_EXIST, tenantCode);
}
putMsg(result, Status.SUCCESS);
return result;
}
@ -317,8 +339,7 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
* @param tenantCode tenant code
* @return ture if the tenant code exists, otherwise return false
*/
@Override
public boolean checkTenantExists(String tenantCode) {
private boolean checkTenantExists(String tenantCode) {
Boolean existTenant = tenantMapper.existTenant(tenantCode);
return Boolean.TRUE.equals(existTenant);
}
@ -339,4 +360,26 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
}
return result;
}
/**
* Make sure tenant with given name exists, and create the tenant if not exists
*
* ONLY for python gateway server, and should not use this in web ui function
*
* @param tenantCode tenant code
* @param desc The description of tenant object
* @param queue The value of queue which current tenant belong
* @param queueName The name of queue which current tenant belong
* @return Tenant object
*/
@Override
public Tenant createTenantIfNotExists(String tenantCode, String desc, String queue, String queueName) {
if (checkTenantExists(tenantCode)) {
return tenantMapper.queryByTenantCode(tenantCode);
}
tenantCodeValid(tenantCode);
Queue newQueue = queueService.createQueueIfNotExists(queue, queueName);
return createObjToDB(tenantCode, desc, newQueue.getId());
}
}

58
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/QueueControllerTest.java
Unescape View File

@ -42,11 +42,14 @@ import org.springframework.util.MultiValueMap;
public class QueueControllerTest extends AbstractControllerTest {
private static final Logger logger = LoggerFactory.getLogger(QueueControllerTest.class);
private static final String QUEUE_CREATE_STRING = "queue1";
private static final String QUEUE_CREATE_NAME = "queue_create";
private static final String QUEUE_MODIFY_NAME = "queue_modify";
private static final String QUEUE_NAME_CREATE_NAME = "queue_name_create";
private static final String QUEUE_NAME_MODIFY_NAME = "queue_name_modify";
private static final String NOT_EXISTS_NAME = "not_exists";
@Test
public void testQueryList() throws Exception {
MvcResult mvcResult = mockMvc.perform(get("/queues/list")
.header(SESSION_ID, sessionId))
.andExpect(status().isOk())
@ -54,13 +57,14 @@ public class QueueControllerTest extends AbstractControllerTest {
.andReturn();
Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
Assert.assertNotNull(result);
Assert.assertEquals(Status.SUCCESS.getCode(),result.getCode().intValue());
logger.info("query list queue return result:{}", mvcResult.getResponse().getContentAsString());
}
@Test
public void testQueryQueueListPaging() throws Exception {
public void testQueryQueueListPagingEmpty() throws Exception {
MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
paramsMap.add("searchVal","");
paramsMap.add("pageNo","1");
@ -73,17 +77,17 @@ public class QueueControllerTest extends AbstractControllerTest {
.andExpect(content().contentType(MediaType.APPLICATION_JSON))
.andReturn();
Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
Assert.assertNotNull(result);
Assert.assertEquals(Status.SUCCESS.getCode(), result.getCode().intValue());
logger.info("query list-paging queue return result:{}", mvcResult.getResponse().getContentAsString());
}
@Test
public void testCreateQueue() throws Exception {
MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
paramsMap.add("queue", QUEUE_CREATE_STRING);
paramsMap.add("queueName","root.queue1");
paramsMap.add("queue", QUEUE_CREATE_NAME);
paramsMap.add("queueName", QUEUE_NAME_CREATE_NAME);
MvcResult mvcResult = mockMvc.perform(post("/queues")
.header(SESSION_ID, sessionId)
@ -92,17 +96,18 @@ public class QueueControllerTest extends AbstractControllerTest {
.andExpect(content().contentType(MediaType.APPLICATION_JSON))
.andReturn();
Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
Assert.assertNotNull(result);
Assert.assertEquals(Status.SUCCESS.getCode(), result.getCode().intValue());
logger.info("create queue return result:{}", mvcResult.getResponse().getContentAsString());
}
@Test
public void testUpdateQueue() throws Exception {
MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
paramsMap.add("id","1");
paramsMap.add("queue","queue2");
paramsMap.add("queueName","root.queue2");
paramsMap.add("queue", QUEUE_MODIFY_NAME);
paramsMap.add("queueName", QUEUE_NAME_MODIFY_NAME);
MvcResult mvcResult = mockMvc.perform(put("/queues/{id}", 1)
.header(SESSION_ID, sessionId)
@ -111,6 +116,8 @@ public class QueueControllerTest extends AbstractControllerTest {
.andExpect(content().contentType(MediaType.APPLICATION_JSON))
.andReturn();
Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
Assert.assertNotNull(result);
Assert.assertEquals(Status.SUCCESS.getCode(),result.getCode().intValue());
logger.info("update queue return result:{}", mvcResult.getResponse().getContentAsString());
}
@ -120,8 +127,8 @@ public class QueueControllerTest extends AbstractControllerTest {
// queue value exist
MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
paramsMap.add("queue",QUEUE_CREATE_STRING);
paramsMap.add("queueName","queue.name");
paramsMap.add("queue", QUEUE_MODIFY_NAME);
paramsMap.add("queueName", NOT_EXISTS_NAME);
MvcResult mvcResult = mockMvc.perform(post("/queues/verify")
.header(SESSION_ID, sessionId)
@ -129,13 +136,32 @@ public class QueueControllerTest extends AbstractControllerTest {
.andExpect(status().isOk())
.andExpect(content().contentType(MediaType.APPLICATION_JSON))
.andReturn();
Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
Assert.assertNotNull(result);
Assert.assertEquals(Status.QUEUE_VALUE_EXIST.getCode(),result.getCode().intValue());
// queue name exist
paramsMap.clear();
paramsMap.add("queue", NOT_EXISTS_NAME);
paramsMap.add("queueName", QUEUE_NAME_CREATE_NAME);
mvcResult = mockMvc.perform(post("/queues/verify")
.header(SESSION_ID, sessionId)
.params(paramsMap))
.andExpect(status().isOk())
.andExpect(content().contentType(MediaType.APPLICATION_JSON))
.andReturn();
result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
Assert.assertNotNull(result);
Assert.assertEquals(Status.QUEUE_NAME_EXIST.getCode(),result.getCode().intValue());
// success
paramsMap.clear();
paramsMap.add("queue","ait123");
paramsMap.add("queueName","aitName");
paramsMap.add("queue", NOT_EXISTS_NAME);
paramsMap.add("queueName", NOT_EXISTS_NAME);
mvcResult = mockMvc.perform(post("/queues/verify")
.header(SESSION_ID, sessionId)
@ -144,6 +170,8 @@ public class QueueControllerTest extends AbstractControllerTest {
.andExpect(content().contentType(MediaType.APPLICATION_JSON))
.andReturn();
result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
Assert.assertNotNull(result);
Assert.assertEquals(Status.SUCCESS.getCode(),result.getCode().intValue());
logger.info(mvcResult.getResponse().getContentAsString());
logger.info("verify queue return result:{}", mvcResult.getResponse().getContentAsString());

144
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/QueueServiceTest.java
Unescape View File

@ -17,7 +17,13 @@
package org.apache.dolphinscheduler.api.service;
import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.YARN_QUEUE_CREATE;
import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.YARN_QUEUE_UPDATE;
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.exceptions.ServiceException;
import org.apache.dolphinscheduler.api.permission.ResourcePermissionCheckService;
import org.apache.dolphinscheduler.api.service.impl.BaseServiceImpl;
import org.apache.dolphinscheduler.api.service.impl.QueueServiceImpl;
import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
@ -30,6 +36,7 @@ import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.commons.collections.CollectionUtils;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@ -38,6 +45,7 @@ import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.jupiter.api.Assertions;
import org.junit.runner.RunWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
@ -55,7 +63,8 @@ import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
@RunWith(MockitoJUnitRunner.class)
public class QueueServiceTest {
private static final Logger logger = LoggerFactory.getLogger(QueueServiceTest.class);
private static final Logger baseServiceLogger = LoggerFactory.getLogger(BaseServiceImpl.class);
private static final Logger queueServiceImplLogger = LoggerFactory.getLogger(QueueServiceImpl.class);
@InjectMocks
private QueueServiceImpl queueService;
@ -66,7 +75,13 @@ public class QueueServiceTest {
@Mock
private UserMapper userMapper;
private String queueName = "QueueServiceTest";
@Mock
private ResourcePermissionCheckService resourcePermissionCheckService;
private static final String QUEUE = "queue";
private static final String QUEUE_NAME = "queueName";
private static final String EXISTS = "exists";
private static final String NOT_EXISTS = "not_exists";
@Before
public void setUp() {
@ -78,10 +93,11 @@ public class QueueServiceTest {
@Test
public void testQueryList() {
Mockito.when(queueMapper.selectList(null)).thenReturn(getQueueList());
Set<Integer> ids = new HashSet<>();
ids.add(1);
Mockito.when(resourcePermissionCheckService.userOwnedResourceIdsAcquisition(AuthorizationType.QUEUE, getLoginUser().getId(), queueServiceImplLogger)).thenReturn(ids);
Mockito.when(queueMapper.selectBatchIds(Mockito.anySet())).thenReturn(getQueueList());
Map<String, Object> result = queueService.queryList(getLoginUser());
logger.info(result.toString());
List<Queue> queueList = (List<Queue>) result.get(Constants.DATA_LIST);
Assert.assertTrue(CollectionUtils.isNotEmpty(queueList));
@ -93,92 +109,96 @@ public class QueueServiceTest {
IPage<Queue> page = new Page<>(1, 10);
page.setTotal(1L);
page.setRecords(getQueueList());
Mockito.when(queueMapper.queryQueuePaging(Mockito.any(Page.class), Mockito.eq(queueName))).thenReturn(page);
Result result = queueService.queryList(getLoginUser(), queueName, 1, 10);
logger.info(result.toString());
Set<Integer> ids = new HashSet<>();
ids.add(1);
Mockito.when(resourcePermissionCheckService.userOwnedResourceIdsAcquisition(AuthorizationType.QUEUE, getLoginUser().getId(), queueServiceImplLogger)).thenReturn(ids);
Mockito.when(queueMapper.queryQueuePaging(Mockito.any(Page.class), Mockito.anyList(), Mockito.eq(QUEUE_NAME))).thenReturn(page);
Result result = queueService.queryList(getLoginUser(), QUEUE_NAME, 1, 10);
PageInfo<Queue> pageInfo = (PageInfo<Queue>) result.getData();
Assert.assertTrue(CollectionUtils.isNotEmpty(pageInfo.getTotalList()));
}
@Test
public void testCreateQueue() {
Mockito.when(resourcePermissionCheckService.operationPermissionCheck(AuthorizationType.QUEUE, getLoginUser().getId(), YARN_QUEUE_CREATE, baseServiceLogger)).thenReturn(true);
Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.QUEUE, null, 0, baseServiceLogger)).thenReturn(true);
// queue is null
Map<String, Object> result = queueService.createQueue(getLoginUser(), null, queueName);
logger.info(result.toString());
Assert.assertEquals(Status.REQUEST_PARAMS_NOT_VALID_ERROR, result.get(Constants.STATUS));
Throwable exception = Assertions.assertThrows(ServiceException.class, () -> queueService.createQueue(getLoginUser(), null, QUEUE_NAME));
String formatter = MessageFormat.format(Status.REQUEST_PARAMS_NOT_VALID_ERROR.getMsg(), Constants.QUEUE);
Assertions.assertEquals(formatter, exception.getMessage());
// queueName is null
result = queueService.createQueue(getLoginUser(), queueName, null);
logger.info(result.toString());
Assert.assertEquals(Status.REQUEST_PARAMS_NOT_VALID_ERROR, result.get(Constants.STATUS));
exception = Assertions.assertThrows(ServiceException.class, () -> queueService.createQueue(getLoginUser(), QUEUE_NAME, null));
formatter = MessageFormat.format(Status.REQUEST_PARAMS_NOT_VALID_ERROR.getMsg(), Constants.QUEUE_NAME);
Assertions.assertEquals(formatter, exception.getMessage());
// correct
result = queueService.createQueue(getLoginUser(), queueName, queueName);
logger.info(result.toString());
Map<String, Object> result = queueService.createQueue(getLoginUser(), QUEUE_NAME, QUEUE_NAME);
Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
}
@Test
public void testUpdateQueue() {
Mockito.when(queueMapper.selectById(1)).thenReturn(getQueue());
Mockito.when(queueMapper.existQueue("test", null)).thenReturn(true);
Mockito.when(queueMapper.existQueue(null, "test")).thenReturn(true);
Mockito.when(queueMapper.selectById(1)).thenReturn(getQUEUE());
Mockito.when(queueMapper.existQueue(EXISTS, null)).thenReturn(true);
Mockito.when(queueMapper.existQueue(null, EXISTS)).thenReturn(true);
Mockito.when(resourcePermissionCheckService.operationPermissionCheck(AuthorizationType.QUEUE, getLoginUser().getId(), YARN_QUEUE_UPDATE, baseServiceLogger)).thenReturn(true);
Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.QUEUE, new Object[]{0}, 0, baseServiceLogger)).thenReturn(true);
// not exist
Map<String, Object> result = queueService.updateQueue(getLoginUser(), 0, "queue", queueName);
logger.info(result.toString());
Assert.assertEquals(Status.QUEUE_NOT_EXIST.getCode(), ((Status) result.get(Constants.STATUS)).getCode());
Throwable exception = Assertions.assertThrows(ServiceException.class, () -> queueService.updateQueue(getLoginUser(), 0, QUEUE, QUEUE_NAME));
String formatter = MessageFormat.format(Status.QUEUE_NOT_EXIST.getMsg(), QUEUE);
Assertions.assertEquals(formatter, exception.getMessage());
//no need update
result = queueService.updateQueue(getLoginUser(), 1, queueName, queueName);
logger.info(result.toString());
Assert.assertEquals(Status.NEED_NOT_UPDATE_QUEUE.getCode(), ((Status) result.get(Constants.STATUS)).getCode());
Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.QUEUE, new Object[]{1}, 0, baseServiceLogger)).thenReturn(true);
exception = Assertions.assertThrows(ServiceException.class, () -> queueService.updateQueue(getLoginUser(), 1, QUEUE_NAME, QUEUE_NAME));
Assertions.assertEquals(Status.NEED_NOT_UPDATE_QUEUE.getMsg(), exception.getMessage());
//queue exist
result = queueService.updateQueue(getLoginUser(), 1, "test", queueName);
logger.info(result.toString());
Assert.assertEquals(Status.QUEUE_VALUE_EXIST.getCode(), ((Status) result.get(Constants.STATUS)).getCode());
exception = Assertions.assertThrows(ServiceException.class, () -> queueService.updateQueue(getLoginUser(), 1, EXISTS, QUEUE_NAME));
formatter = MessageFormat.format(Status.QUEUE_VALUE_EXIST.getMsg(), EXISTS);
Assertions.assertEquals(formatter, exception.getMessage());
// queueName exist
result = queueService.updateQueue(getLoginUser(), 1, "test1", "test");
logger.info(result.toString());
Assert.assertEquals(Status.QUEUE_NAME_EXIST.getCode(), ((Status) result.get(Constants.STATUS)).getCode());
exception = Assertions.assertThrows(ServiceException.class, () -> queueService.updateQueue(getLoginUser(), 1, NOT_EXISTS, EXISTS));
formatter = MessageFormat.format(Status.QUEUE_NAME_EXIST.getMsg(), EXISTS);
Assertions.assertEquals(formatter, exception.getMessage());
//success
result = queueService.updateQueue(getLoginUser(), 1, "test1", "test1");
logger.info(result.toString());
Mockito.when(userMapper.existUser(Mockito.anyString())).thenReturn(false);
Map<String, Object> result = queueService.updateQueue(getLoginUser(), 1, NOT_EXISTS, NOT_EXISTS);
Assert.assertEquals(Status.SUCCESS.getCode(), ((Status) result.get(Constants.STATUS)).getCode());
}
@Test
public void testVerifyQueue() {
Mockito.when(queueMapper.existQueue(queueName, null)).thenReturn(true);
Mockito.when(queueMapper.existQueue(null, queueName)).thenReturn(true);
//queue null
Result result = queueService.verifyQueue(null, queueName);
logger.info(result.toString());
Assert.assertEquals(result.getCode().intValue(), Status.REQUEST_PARAMS_NOT_VALID_ERROR.getCode());
Throwable exception = Assertions.assertThrows(ServiceException.class, () -> queueService.verifyQueue(null, QUEUE_NAME));
String formatter = MessageFormat.format(Status.REQUEST_PARAMS_NOT_VALID_ERROR.getMsg(), Constants.QUEUE);
Assertions.assertEquals(formatter, exception.getMessage());
//queueName null
result = queueService.verifyQueue(queueName, null);
logger.info(result.toString());
Assert.assertEquals(result.getCode().intValue(), Status.REQUEST_PARAMS_NOT_VALID_ERROR.getCode());
exception = Assertions.assertThrows(ServiceException.class, () -> queueService.verifyQueue(QUEUE_NAME, null));
formatter = MessageFormat.format(Status.REQUEST_PARAMS_NOT_VALID_ERROR.getMsg(), Constants.QUEUE_NAME);
Assertions.assertEquals(formatter, exception.getMessage());
//exist queueName
result = queueService.verifyQueue(queueName, queueName);
logger.info(result.toString());
Assert.assertEquals(result.getCode().intValue(), Status.QUEUE_NAME_EXIST.getCode());
Mockito.when(queueMapper.existQueue(EXISTS, null)).thenReturn(true);
exception = Assertions.assertThrows(ServiceException.class, () -> queueService.verifyQueue(EXISTS, QUEUE_NAME));
formatter = MessageFormat.format(Status.QUEUE_VALUE_EXIST.getMsg(), EXISTS);
Assertions.assertEquals(formatter, exception.getMessage());
//exist queue
result = queueService.verifyQueue(queueName, "test");
logger.info(result.toString());
Assert.assertEquals(result.getCode().intValue(), Status.QUEUE_VALUE_EXIST.getCode());
Mockito.when(queueMapper.existQueue(null, EXISTS)).thenReturn(true);
exception = Assertions.assertThrows(ServiceException.class, () -> queueService.verifyQueue(QUEUE, EXISTS));
formatter = MessageFormat.format(Status.QUEUE_NAME_EXIST.getMsg(), EXISTS);
Assertions.assertEquals(formatter, exception.getMessage());
// success
result = queueService.verifyQueue("test", "test");
logger.info(result.toString());
Result<Object> result = queueService.verifyQueue(NOT_EXISTS, NOT_EXISTS);
Assert.assertEquals(result.getCode().intValue(), Status.SUCCESS.getCode());
}
/**
@ -192,26 +212,20 @@ public class QueueServiceTest {
return loginUser;
}
private List<User> getUserList() {
List<User> list = new ArrayList<>();
list.add(getLoginUser());
return list;
}
/**
* get queue
*/
private Queue getQueue() {
private Queue getQUEUE() {
Queue queue = new Queue();
queue.setId(1);
queue.setQueue(queueName);
queue.setQueueName(queueName);
queue.setQueue(QUEUE_NAME);
queue.setQueueName(QUEUE_NAME);
return queue;
}
private List<Queue> getQueueList() {
List<Queue> queueList = new ArrayList<>();
queueList.add(getQueue());
queueList.add(getQUEUE());
return queueList;
}

169
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TenantServiceTest.java
Unescape View File

@ -17,16 +17,20 @@
package org.apache.dolphinscheduler.api.service;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import org.apache.commons.collections.CollectionUtils;
import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.TENANT_CREATE;
import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.TENANT_DELETE;
import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.TENANT_UPDATE;
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.exceptions.ServiceException;
import org.apache.dolphinscheduler.api.permission.ResourcePermissionCheckService;
import org.apache.dolphinscheduler.api.service.impl.BaseServiceImpl;
import org.apache.dolphinscheduler.api.service.impl.TenantServiceImpl;
import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.AuthorizationType;
import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.common.storage.StorageOperate;
import org.apache.dolphinscheduler.common.utils.PropertyUtils;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
@ -36,8 +40,19 @@ import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.commons.collections.CollectionUtils;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.junit.Assert;
import org.junit.Test;
import org.junit.jupiter.api.Assertions;
import org.junit.runner.RunWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
@ -47,9 +62,8 @@ import org.powermock.core.classloader.annotations.PrepareForTest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
/**
* tenant service test
@ -57,8 +71,8 @@ import java.util.Map;
@RunWith(MockitoJUnitRunner.class)
@PrepareForTest({PropertyUtils.class})
public class TenantServiceTest {
private static final Logger logger = LoggerFactory.getLogger(TenantServiceTest.class);
private static final Logger baseServiceLogger = LoggerFactory.getLogger(BaseServiceImpl.class);
private static final Logger tenantServiceImplLogger = LoggerFactory.getLogger(TenantServiceImpl.class);
@InjectMocks
private TenantServiceImpl tenantService;
@ -76,136 +90,143 @@ public class TenantServiceTest {
private UserMapper userMapper;
@Mock
private StorageOperate storageOperate;
private ResourcePermissionCheckService resourcePermissionCheckService;
private static final String tenantCode = "hayden";
private static final String tenantDesc = "This is the tenant desc";
@Test
public void testCreateTenant() {
public void testCreateTenant() throws Exception {
User loginUser = getLoginUser();
Mockito.when(tenantMapper.existTenant(tenantCode)).thenReturn(true);
try {
//check tenantCode
Map<String, Object> result =
tenantService.createTenant(getLoginUser(), "%!1111", 1, "TenantServiceTest");
logger.info(result.toString());
Assert.assertEquals(Status.CHECK_OS_TENANT_CODE_ERROR, result.get(Constants.STATUS));
Mockito.when(resourcePermissionCheckService.operationPermissionCheck(AuthorizationType.TENANT, loginUser.getId(), TENANT_CREATE, baseServiceLogger)).thenReturn(true);
Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.TENANT, null, 0, baseServiceLogger)).thenReturn(true);
Map<String, Object> result;
//check exist
result = tenantService.createTenant(loginUser, tenantCode, 1, "TenantServiceTest");
logger.info(result.toString());
Assert.assertEquals(Status.OS_TENANT_CODE_EXIST, result.get(Constants.STATUS));
String emptyTenantCode = "";
Throwable exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.createTenant(loginUser, emptyTenantCode, 1, tenantDesc));
String formatter = MessageFormat.format(Status.REQUEST_PARAMS_NOT_VALID_ERROR.getMsg(), emptyTenantCode);
Assertions.assertEquals(formatter, exception.getMessage());
//check tenant code too long
String longStr = "this_is_a_very_long_string_this_is_a_very_long_string_this_is_a_very_long_string_this_is_a_very_long_string";
exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.createTenant(loginUser, longStr, 1, tenantDesc));
Assert.assertEquals(Status.TENANT_FULL_NAME_TOO_LONG_ERROR.getMsg(), exception.getMessage());
//check tenant code invalid
exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.createTenant(getLoginUser(), "%!1111", 1, tenantDesc));
Assert.assertEquals(Status.CHECK_OS_TENANT_CODE_ERROR.getMsg(), exception.getMessage());
//check exist
exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.createTenant(loginUser, tenantCode, 1, tenantDesc));
formatter = MessageFormat.format(Status.OS_TENANT_CODE_EXIST.getMsg(), tenantCode);
Assert.assertEquals(formatter, exception.getMessage());
// success
result = tenantService.createTenant(loginUser, "test", 1, "TenantServiceTest");
logger.info(result.toString());
result = tenantService.createTenant(loginUser, "test", 1, tenantDesc);
Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
} catch (Exception e) {
logger.error("create tenant error", e);
Assert.fail();
}
@Test
public void testCreateTenantError() {
Mockito.when(tenantMapper.existTenant(tenantCode)).thenReturn(true);
// tenantCode exist
Throwable exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.verifyTenantCode(getTenant().getTenantCode()));
String expect = MessageFormat.format(Status.OS_TENANT_CODE_EXIST.getMsg(), getTenant().getTenantCode());
Assertions.assertEquals(expect, exception.getMessage());
// success
Result result = tenantService.verifyTenantCode("s00000000000l887888885554444sfjdskfjslakslkdf");
Assert.assertEquals(Status.SUCCESS.getMsg(), result.getMsg());
}
@Test
@SuppressWarnings("unchecked")
public void testQueryTenantListPage() {
IPage<Tenant> page = new Page<>(1, 10);
page.setRecords(getList());
page.setTotal(1L);
Mockito.when(tenantMapper.queryTenantPaging(Mockito.any(Page.class), Mockito.eq("TenantServiceTest")))
Set<Integer> ids = new HashSet<>();
ids.add(1);
Mockito.when(resourcePermissionCheckService.userOwnedResourceIdsAcquisition(AuthorizationType.TENANT, getLoginUser().getId(), tenantServiceImplLogger)).thenReturn(ids);
Mockito.when(tenantMapper.queryTenantPaging(Mockito.any(Page.class), Mockito.anyList(), Mockito.eq(tenantDesc)))
.thenReturn(page);
Result result = tenantService.queryTenantList(getLoginUser(), "TenantServiceTest", 1, 10);
logger.info(result.toString());
Result result = tenantService.queryTenantList(getLoginUser(), tenantDesc, 1, 10);
PageInfo<Tenant> pageInfo = (PageInfo<Tenant>) result.getData();
Assert.assertTrue(CollectionUtils.isNotEmpty(pageInfo.getTotalList()));
}
@Test
public void testUpdateTenant() {
public void testUpdateTenant() throws Exception {
Mockito.when(tenantMapper.queryById(1)).thenReturn(getTenant());
try {
// id not exist
Map<String, Object> result =
tenantService.updateTenant(getLoginUser(), 912222, tenantCode, 1, "desc");
logger.info(result.toString());
// success
Assert.assertEquals(Status.TENANT_NOT_EXIST, result.get(Constants.STATUS));
result = tenantService.updateTenant(getLoginUser(), 1, tenantCode, 1, "desc");
logger.info(result.toString());
Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
} catch (Exception e) {
logger.error("update tenant error", e);
Assert.fail();
}
Mockito.when(resourcePermissionCheckService.operationPermissionCheck(AuthorizationType.TENANT, getLoginUser().getId(), TENANT_UPDATE, baseServiceLogger)).thenReturn(true);
Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.TENANT, null, 0, baseServiceLogger)).thenReturn(true);
Throwable exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.updateTenant(getLoginUser(), 912222, tenantCode, 1, tenantDesc));
Assertions.assertEquals(Status.TENANT_NOT_EXIST.getMsg(), exception.getMessage());
Map<String, Object> result = tenantService.updateTenant(getLoginUser(), 1, tenantCode, 1, tenantDesc);
Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
}
@Test
public void testDeleteTenantById() {
public void testDeleteTenantById() throws Exception {
Mockito.when(resourcePermissionCheckService.operationPermissionCheck(AuthorizationType.TENANT, getLoginUser().getId(), TENANT_DELETE, baseServiceLogger)).thenReturn(true);
Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.TENANT, null, 0, baseServiceLogger)).thenReturn(true);
Mockito.when(tenantMapper.queryById(1)).thenReturn(getTenant());
Mockito.when(processInstanceMapper.queryByTenantIdAndStatus(1, Constants.NOT_TERMINATED_STATES))
.thenReturn(getInstanceList());
Mockito.when(processDefinitionMapper.queryDefinitionListByTenant(2)).thenReturn(getDefinitionsList());
Mockito.when(userMapper.queryUserListByTenant(3)).thenReturn(getUserList());
try {
//TENANT_NOT_EXIST
Map<String, Object> result = tenantService.deleteTenantById(getLoginUser(), 12);
logger.info(result.toString());
Assert.assertEquals(Status.TENANT_NOT_EXIST, result.get(Constants.STATUS));
Throwable exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.deleteTenantById(getLoginUser(), 12));
Assertions.assertEquals(Status.TENANT_NOT_EXIST.getMsg(), exception.getMessage());
//DELETE_TENANT_BY_ID_FAIL
result = tenantService.deleteTenantById(getLoginUser(), 1);
logger.info(result.toString());
Assert.assertEquals(Status.DELETE_TENANT_BY_ID_FAIL, result.get(Constants.STATUS));
exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.deleteTenantById(getLoginUser(), 1));
String prefix = Status.DELETE_TENANT_BY_ID_FAIL.getMsg().substring(1, 5);
Assertions.assertTrue(exception.getMessage().contains(prefix));
//DELETE_TENANT_BY_ID_FAIL_DEFINES
Mockito.when(tenantMapper.queryById(2)).thenReturn(getTenant(2));
result = tenantService.deleteTenantById(getLoginUser(), 2);
logger.info(result.toString());
Assert.assertEquals(Status.DELETE_TENANT_BY_ID_FAIL_DEFINES, result.get(Constants.STATUS));
exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.deleteTenantById(getLoginUser(), 2));
prefix = Status.DELETE_TENANT_BY_ID_FAIL_DEFINES.getMsg().substring(1, 5);
Assertions.assertTrue(exception.getMessage().contains(prefix));
//DELETE_TENANT_BY_ID_FAIL_USERS
Mockito.when(tenantMapper.queryById(3)).thenReturn(getTenant(3));
result = tenantService.deleteTenantById(getLoginUser(), 3);
logger.info(result.toString());
Assert.assertEquals(Status.DELETE_TENANT_BY_ID_FAIL_USERS, result.get(Constants.STATUS));
exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.deleteTenantById(getLoginUser(), 3));
prefix = Status.DELETE_TENANT_BY_ID_FAIL_USERS.getMsg().substring(1, 5);
Assertions.assertTrue(exception.getMessage().contains(prefix));
// success
Mockito.when(tenantMapper.queryById(4)).thenReturn(getTenant(4));
result = tenantService.deleteTenantById(getLoginUser(), 4);
logger.info(result.toString());
Map<String, Object> result = tenantService.deleteTenantById(getLoginUser(), 4);
Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
} catch (Exception e) {
logger.error("delete tenant error", e);
Assert.fail();
}
}
@Test
public void testVerifyTenantCode() {
Mockito.when(tenantMapper.existTenant(tenantCode)).thenReturn(true);
// tenantCode not exist
// tenantCode exist
Throwable exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.verifyTenantCode(getTenant().getTenantCode()));
String expect = MessageFormat.format(Status.OS_TENANT_CODE_EXIST.getMsg(), getTenant().getTenantCode());
Assertions.assertEquals(expect, exception.getMessage());
// success
Result result = tenantService.verifyTenantCode("s00000000000l887888885554444sfjdskfjslakslkdf");
logger.info(result.toString());
Assert.assertEquals(Status.SUCCESS.getMsg(), result.getMsg());
// tenantCode exist
result = tenantService.verifyTenantCode(getTenant().getTenantCode());
Assert.assertEquals(Status.OS_TENANT_CODE_EXIST.getCode(), result.getCode().intValue());
}
/**
* get user
*/
private User getLoginUser() {
User loginUser = new User();
loginUser.setUserType(UserType.ADMIN_USER);
return loginUser;

7
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/QueueMapper.java
Unescape View File

@ -55,9 +55,10 @@ public interface QueueMapper extends BaseMapper<Queue> {
Boolean existQueue(@Param("queue") String queue, @Param("queueName") String queueName);
/**
* query queue by queue name
* query simple queue object by queue name and queue
* @param queue queue
* @param queueName queueName
* @return queue list
* @return queue object
*/
List<Queue> queryQueueName(@Param("queueName") String queueName);
Queue queryQueueName(@Param("queue") String queue, @Param("queueName") String queueName);
}

6
dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/QueueMapper.xml
Unescape View File

@ -58,9 +58,13 @@
select
<include refid="baseSql"/>
from t_ds_queue
where 1 = 1
<where>
<if test="queue != null and queue != ''">
and queue =#{queue}
</if>
<if test="queueName != null and queueName != ''">
and queue_name =#{queueName}
</if>
</where>
</select>
</mapper>

Write Preview
Loading…
Cancel
Save