ITests for Stream Consumer Groups functionality

redisinsight/api/src/constants/error-messages.ts

@@ -3,6 +3,7 @@ export default {
   INVALID_DATABASE_INSTANCE_ID: 'Invalid database instance id.',
   COMMAND_EXECUTION_NOT_FOUND: 'Command execution was not found.',
   PROFILER_LOG_FILE_NOT_FOUND: 'Profiler log file was not found.',
+  CONSUMER_GROUP_NOT_FOUND: 'Consumer Group with such name was not found.',
   PLUGIN_STATE_NOT_FOUND: 'Plugin state was not found.',
   UNDEFINED_INSTANCE_ID: 'Undefined redis database instance id.',
   NO_CONNECTION_TO_REDIS_DB: 'No connection to the Redis Database.',

redisinsight/api/src/constants/redis-error-codes.ts

@@ -11,6 +11,7 @@ export enum RedisErrorCodes {
   Timeout = 'ETIMEDOUT',
   CommandSyntaxError = 'syntax error',
   BusyGroup = 'BUSYGROUP',
+  NoGroup = 'NOGROUP',
   UnknownCommand = 'unknown command',
 }
 

redisinsight/api/src/modules/browser/dto/stream.dto.ts

@@ -5,7 +5,16 @@ import {
   ArrayNotEmpty,
   IsArray,
   IsDefined,
-  IsEnum, IsInt, IsNotEmpty, IsString, Min, ValidateNested, isString, IsOptional, NotEquals, ValidateIf, IsBoolean,
+  IsEnum,
+  IsInt,
+  IsNotEmpty,
+  IsString,
+  Min,
+  ValidateNested,
+  isString,
+  NotEquals,
+  ValidateIf,
+  IsBoolean,
 } from 'class-validator';
 import { KeyDto, KeyWithExpireDto } from 'src/modules/browser/dto/keys.dto';
 import { SortOrder } from 'src/constants';
@@ -278,7 +287,8 @@ export class DeleteConsumerGroupsDto extends KeyDto {
   @IsDefined()
   @IsArray()
   @ArrayNotEmpty()
-  @Type(() => String)
+  @IsNotEmpty({ each: true })
+  @IsString({ each: true })
   consumerGroups: string[];
 }
 
@@ -335,7 +345,8 @@ export class DeleteConsumersDto extends GetConsumersDto {
   @IsDefined()
   @IsArray()
   @ArrayNotEmpty()
-  @Type(() => String)
+  @IsString({ each: true })
+  @IsNotEmpty({ each: true })
   consumerNames: string[];
 }
 
@@ -421,7 +432,8 @@ export class AckPendingEntriesDto extends GetConsumersDto {
   @IsDefined()
   @IsArray()
   @ArrayNotEmpty()
-  @Type(() => String)
+  @IsString({ each: true })
+  @IsNotEmpty({ each: true })
   entries: string[];
 }
 
@@ -471,7 +483,8 @@ export class ClaimPendingEntryDto extends KeyDto {
   @IsDefined()
   @IsArray()
   @ArrayNotEmpty()
-  @Type(() => String)
+  @IsString({ each: true })
+  @IsNotEmpty({ each: true })
   entries: string[];
 
   @ApiPropertyOptional({

redisinsight/api/src/modules/browser/services/stream/consumer-group.service.ts

@@ -15,7 +15,7 @@ import {
   ConsumerGroupDto,
   CreateConsumerGroupsDto,
   DeleteConsumerGroupsDto, DeleteConsumerGroupsResponse,
-  UpdateConsumerGroupDto
+  UpdateConsumerGroupDto,
 } from 'src/modules/browser/dto/stream.dto';
 
 @Injectable()
@@ -197,6 +197,10 @@ export class ConsumerGroupService {
         throw new BadRequestException(error.message);
       }
 
+      if (error?.message.includes(RedisErrorCodes.NoGroup)) {
+        throw new NotFoundException(ERROR_MESSAGES.CONSUMER_GROUP_NOT_FOUND);
+      }
+
       throw catchAclError(error);
     }
   }

redisinsight/api/src/modules/browser/services/stream/consumer.service.ts

@@ -3,7 +3,7 @@ import {
 } from '@nestjs/common';
 import { IFindRedisClientInstanceByOptions } from 'src/modules/core/services/redis/redis.service';
 import { RedisErrorCodes } from 'src/constants';
-import {catchAclError, catchTransactionError, convertStringsArrayToObject} from 'src/utils';
+import { catchAclError, catchTransactionError, convertStringsArrayToObject } from 'src/utils';
 import {
   BrowserToolCommands,
   BrowserToolKeysCommands, BrowserToolStreamCommands,
@@ -54,6 +54,10 @@ export class ConsumerService {
         throw error;
       }
 
+      if (error?.message.includes(RedisErrorCodes.NoGroup)) {
+        throw new NotFoundException(ERROR_MESSAGES.CONSUMER_GROUP_NOT_FOUND);
+      }
+
       if (error?.message.includes(RedisErrorCodes.WrongType)) {
         throw new BadRequestException(error.message);
       }
@@ -108,6 +112,10 @@ export class ConsumerService {
         throw error;
       }
 
+      if (error?.message.includes(RedisErrorCodes.NoGroup)) {
+        throw new NotFoundException(ERROR_MESSAGES.CONSUMER_GROUP_NOT_FOUND);
+      }
+
       if (error?.message.includes(RedisErrorCodes.WrongType)) {
         throw new BadRequestException(error.message);
       }
@@ -148,6 +156,10 @@ export class ConsumerService {
         throw error;
       }
 
+      if (error?.message.includes(RedisErrorCodes.NoGroup)) {
+        throw new NotFoundException(ERROR_MESSAGES.CONSUMER_GROUP_NOT_FOUND);
+      }
+
       if (error?.message.includes(RedisErrorCodes.WrongType)) {
         throw new BadRequestException(error.message);
       }
@@ -259,6 +271,10 @@ export class ConsumerService {
         throw error;
       }
 
+      if (error?.message.includes(RedisErrorCodes.NoGroup)) {
+        throw new NotFoundException(ERROR_MESSAGES.CONSUMER_GROUP_NOT_FOUND);
+      }
+
       if (error?.message.includes(RedisErrorCodes.WrongType)) {
         throw new BadRequestException(error.message);
       }

redisinsight/api/test/api/stream/DELETE-instance-id-streams-consumer_groups-consumers.test.ts

@@ -0,0 +1,232 @@
+import {
+  expect,
+  describe,
+  it,
+  before,
+  deps,
+  Joi,
+  requirements,
+  generateInvalidDataTestCases,
+  validateInvalidDataTestCase,
+  validateApiCall
+} from '../deps';
+const { server, request, constants, rte } = deps;
+
+// endpoint to test
+const endpoint = (instanceId = constants.TEST_INSTANCE_ID) =>
+  request(server).delete(`/instance/${instanceId}/streams/consumer-groups/consumers`);
+
+const dataSchema = Joi.object({
+  keyName: Joi.string().allow('').required(),
+  groupName: Joi.string().required().messages({
+    'any.required': '{#label} should not be empty',
+  }),
+  consumerNames: Joi.array().items(Joi.string().required().label('consumerNames').messages({
+    'any.required': '{#label} should not be empty',
+  })).min(1).required().messages({
+    'array.sparse': 'each value in consumerNames must be a string',
+  }),
+}).strict();
+
+const validInputData = {
+  keyName: constants.TEST_STREAM_KEY_1,
+  groupName: constants.TEST_STREAM_GROUP_1,
+  consumerNames: [constants.TEST_STREAM_GROUP_1],
+};
+
+
+const mainCheckFn = async (testCase) => {
+  it(testCase.name, async () => {
+    // additional checks before test run
+    if (testCase.before) {
+      await testCase.before();
+    }
+
+    await validateApiCall({
+      endpoint,
+      ...testCase,
+    });
+
+    // additional checks after test pass
+    if (testCase.after) {
+      await testCase.after();
+    }
+  });
+};
+
+describe('DELETE /instance/:instanceId/streams/consumer-groups/consumers', () => {
+  before(async () => await rte.data.generateKeys(true));
+
+  describe('Validation', () => {
+    generateInvalidDataTestCases(dataSchema, validInputData).map(
+      validateInvalidDataTestCase(endpoint, dataSchema),
+    );
+  });
+
+  describe('Common', () => {
+    beforeEach(async () => {
+      await rte.data.sendCommand('xreadgroup', [
+        'GROUP',
+        constants.TEST_STREAM_GROUP_1,
+        constants.TEST_STREAM_CONSUMER_1,
+        'STREAMS',
+        constants.TEST_STREAM_KEY_1,
+        constants.TEST_STREAM_ID_1,
+      ]);
+      await rte.data.sendCommand('xreadgroup', [
+        'GROUP',
+        constants.TEST_STREAM_GROUP_1,
+        constants.TEST_STREAM_CONSUMER_2,
+        'STREAMS',
+        constants.TEST_STREAM_KEY_1,
+        constants.TEST_STREAM_ID_2,
+      ]);
+    });
+
+    [
+      {
+        name: 'Should remove single consumer',
+        data: {
+          keyName: constants.TEST_STREAM_KEY_1,
+          groupName: constants.TEST_STREAM_GROUP_1,
+          consumerNames: [constants.TEST_STREAM_CONSUMER_1],
+        },
+        before: async () => {
+          const consumers = await rte.data.sendCommand('xinfo', ['consumers', constants.TEST_STREAM_KEY_1, constants.TEST_STREAM_GROUP_1]);
+          expect(consumers.length).to.eq(2);
+        },
+        after: async () => {
+          const consumers = await rte.data.sendCommand('xinfo', ['consumers', constants.TEST_STREAM_KEY_1, constants.TEST_STREAM_GROUP_1]);
+          expect(consumers.length).to.eq(1);
+        },
+      },
+      {
+        name: 'Should remove multiple consumers',
+        data: {
+          keyName: constants.TEST_STREAM_KEY_1,
+          groupName: constants.TEST_STREAM_GROUP_1,
+          consumerNames: [constants.TEST_STREAM_CONSUMER_1, constants.TEST_STREAM_CONSUMER_2],
+        },
+        before: async () => {
+          const consumers = await rte.data.sendCommand('xinfo', ['consumers', constants.TEST_STREAM_KEY_1, constants.TEST_STREAM_GROUP_1]);
+          expect(consumers.length).to.eq(2);
+        },
+        after: async () => {
+          const consumers = await rte.data.sendCommand('xinfo', ['consumers', constants.TEST_STREAM_KEY_1, constants.TEST_STREAM_GROUP_1]);
+          expect(consumers.length).to.eq(0);
+        },
+      },
+      {
+        name: 'Should remove single consumers and skip not existing consumers',
+        data: {
+          keyName: constants.TEST_STREAM_KEY_1,
+          groupName: constants.TEST_STREAM_GROUP_1,
+          consumerNames: [constants.TEST_STREAM_CONSUMER_1, constants.getRandomString(), constants.getRandomString()],
+        },
+        before: async () => {
+          const consumers = await rte.data.sendCommand('xinfo', ['consumers', constants.TEST_STREAM_KEY_1, constants.TEST_STREAM_GROUP_1]);
+          console.log('_c', consumers)
+          expect(consumers.length).to.eq(2);
+        },
+        after: async () => {
+          const consumers = await rte.data.sendCommand('xinfo', ['consumers', constants.TEST_STREAM_KEY_1, constants.TEST_STREAM_GROUP_1]);
+          expect(consumers.length).to.eq(1);
+        },
+      },
+      {
+        name: 'Should return BadRequest error if key has another type',
+        data: {
+          ...validInputData,
+          keyName: constants.TEST_STRING_KEY_1,
+        },
+        statusCode: 400,
+        responseBody: {
+          statusCode: 400,
+          error: 'Bad Request',
+        },
+      },
+      {
+        name: 'Should return NotFound error if group does not exists',
+        data: {
+          ...validInputData,
+          groupName: constants.getRandomString(),
+        },
+        statusCode: 404,
+        responseBody: {
+          statusCode: 404,
+          error: 'Not Found',
+          message: 'Consumer Group with such name was not found.',
+        },
+      },
+      {
+        name: 'Should return NotFound error if key does not exists',
+        data: {
+          ...validInputData,
+          keyName: constants.getRandomString(),
+        },
+        statusCode: 404,
+        responseBody: {
+          statusCode: 404,
+          error: 'Not Found',
+          message: 'Key with this name does not exist.',
+        },
+      },
+      {
+        name: 'Should return NotFound error if instance id does not exists',
+        endpoint: () => endpoint(constants.TEST_NOT_EXISTED_INSTANCE_ID),
+        data: {
+          ...validInputData,
+        },
+        statusCode: 404,
+        responseBody: {
+          statusCode: 404,
+          error: 'Not Found',
+          message: 'Invalid database instance id.',
+        },
+      },
+    ].map(mainCheckFn);
+  });
+
+  describe('ACL', () => {
+    requirements('rte.acl');
+
+    before(async () => await rte.data.generateKeys(true));
+    before(async () => rte.data.setAclUserRules('~* +@all'));
+
+    [
+      {
+        name: 'Should create consumer group',
+        endpoint: () => endpoint(constants.TEST_INSTANCE_ACL_ID),
+        data: {
+          ...validInputData,
+        },
+      },
+      {
+        name: 'Should throw error if no permissions for "exists" command',
+        endpoint: () => endpoint(constants.TEST_INSTANCE_ACL_ID),
+        data: {
+          ...validInputData,
+        },
+        statusCode: 403,
+        responseBody: {
+          statusCode: 403,
+          error: 'Forbidden',
+        },
+        before: () => rte.data.setAclUserRules('~* +@all -exists')
+      },
+      {
+        name: 'Should throw error if no permissions for "xgroup)" command',
+        endpoint: () => endpoint(constants.TEST_INSTANCE_ACL_ID),
+        data: {
+          ...validInputData,
+        },
+        statusCode: 403,
+        responseBody: {
+          statusCode: 403,
+          error: 'Forbidden',
+        },
+        before: () => rte.data.setAclUserRules('~* +@all -xgroup')
+      },
+    ].map(mainCheckFn);
+  });
+});