Integration tests for commands, delete operations, expiry, compatibility

Add test for bloom command arity, behavior and basic error

Set up a base case for valkey bloom filter module

Signed-off-by: Vanessa Tang <yuetan@amazon.com>

Optimize BF.INSERT arguments handling and BF.INFO response when NONSCALING set

Signed-off-by: Vanessa Tang <yuetan@amazon.com>

Add test for basic valkey command

Signed-off-by: Vanessa Tang <yuetan@amazon.com>

src/bloom/command_handler.rs

@@ -332,11 +332,17 @@ pub fn bloom_filter_insert(ctx: &Context, input_args: &[ValkeyString]) -> Valkey
                 break;
             }
             _ => {
-                return Err(ValkeyError::WrongArity);
+                return Err(ValkeyError::Str(utils::UNKNOWN_ARGUMENT));
             }
         }
         idx += 1;
     }
+
+    if idx == argc {
+        // No ITEMS argument from the insert command
+        return Err(ValkeyError::WrongArity);
+    }
+
     // If the filter does not exist, create one
     let filter_key = ctx.open_key_writable(filter_name);
     let value = match filter_key.get_value::<BloomFilterType>(&BLOOM_FILTER_TYPE) {
@@ -399,7 +405,7 @@ pub fn bloom_filter_info(ctx: &Context, input_args: &[ValkeyString]) -> ValkeyRe
                 "ITEMS" => Ok(ValkeyValue::Integer(val.cardinality())),
                 "EXPANSION" => {
                     if val.expansion == 0 {
-                        return Ok(ValkeyValue::Integer(-1));
+                        return Ok(ValkeyValue::Null);
                     }
                     Ok(ValkeyValue::Integer(val.expansion as i64))
                 }
@@ -419,7 +425,7 @@ pub fn bloom_filter_info(ctx: &Context, input_args: &[ValkeyString]) -> ValkeyRe
                 ValkeyValue::SimpleStringStatic("Expansion rate"),
             ];
             if val.expansion == 0 {
-                result.push(ValkeyValue::Integer(-1));
+                result.push(ValkeyValue::Null);
             } else {
                 result.push(ValkeyValue::Integer(val.expansion as i64));
             }

src/bloom/utils.rs

@@ -10,13 +10,14 @@ pub const ERROR: &str = "ERROR";
 pub const NON_SCALING_FILTER_FULL: &str = "ERR non scaling filter is full";
 pub const NOT_FOUND: &str = "ERR not found";
 pub const ITEM_EXISTS: &str = "ERR item exists";
-pub const INVALID_INFO_VALUE: &str = "ERR Invalid information value";
+pub const INVALID_INFO_VALUE: &str = "ERR invalid information value";
 pub const BAD_EXPANSION: &str = "ERR bad expansion";
 pub const BAD_CAPACITY: &str = "ERR bad capacity";
 pub const BAD_ERROR_RATE: &str = "ERR bad error rate";
 pub const ERROR_RATE_RANGE: &str = "ERR (0 < error rate range < 1)";
 pub const CAPACITY_LARGER_THAN_0: &str = "ERR (capacity should be larger than 0)";
 pub const MAX_NUM_SCALING_FILTERS: &str = "ERR max number of scaling filters reached";
+pub const UNKNOWN_ARGUMENT: &str = "ERR unknown argument received";
 
 #[derive(Debug, PartialEq)]
 pub enum BloomError {

tests/test_basic.py

@@ -1,18 +1,13 @@
 import time
 import pytest
+from util.waiters import *
 from valkey import ResponseError
-from valkeytests.valkey_test_case import ValkeyTestCase
+from valkey_bloom_test_case import ValkeyBloomTestCaseBase
 from valkeytests.conftest import resource_port_tracker
 import logging
 import os
 
-class TestBloomBasic(ValkeyTestCase):
-
-    def get_custom_args(self):
-        self.set_server_version(os.environ['SERVER_VERSION'])
-        return {
-            'loadmodule': os.getenv('MODULE_PATH'),
-        }
+class TestBloomBasic(ValkeyBloomTestCaseBase):
 
     def test_basic(self):
         client = self.server.get_new_client()
@@ -31,3 +26,98 @@ def test_basic(self):
         assert bf_exists_result == 1
         bf_exists_result = client.execute_command('BF.EXISTS filter1 item2')
         assert bf_exists_result == 0
+
+    def test_bloom_modification(self):
+        client = self.server.get_new_client()
+        # check bloom filter with basic valkey command
+        # cmd touch
+        assert client.execute_command('BF.ADD key1 val1') == 1
+        assert client.execute_command('BF.ADD key2 val2') == 1
+        assert client.execute_command('TOUCH key1 key2') == 2
+        assert client.execute_command('TOUCH key3') == 0
+        self.verify_key_number(client, 2)
+
+    def test_bloom_transaction(self):
+        client = self.server.get_new_client()
+        # cmd multi, exec
+        assert client.execute_command('MULTI') == b'OK'
+        assert client.execute_command('BF.ADD M1 V1') == b'QUEUED'
+        assert client.execute_command('BF.ADD M2 V2') == b'QUEUED'
+        assert client.execute_command('BF.EXISTS M1 V1') == b'QUEUED'
+        assert client.execute_command('DEL M1') == b'QUEUED'
+        assert client.execute_command('BF.EXISTS M1 V1') == b'QUEUED'
+        assert client.execute_command('EXEC') == [1, 1, 1, 1, 0]
+        self.verify_bloom_filter_existence(client, 'M2', 'V2')
+        self.verify_bloom_filter_existence(client, 'M1', 'V1', should_exist=False)
+        self.verify_key_number(client, 1)
+
+    def test_bloom_lua(self):
+        client = self.server.get_new_client()
+        # lua
+        load_filter = """
+        redis.call('BF.ADD', 'LUA1', 'ITEM1');
+        redis.call('BF.ADD', 'LUA2', 'ITEM2');
+        redis.call('BF.MADD', 'LUA2', 'ITEM3', 'ITEM4', 'ITEM5');
+        """
+        client.eval(load_filter, 0)
+        assert client.execute_command('BF.MEXISTS LUA2 ITEM1 ITEM3 ITEM4') == [0, 1, 1]
+        self.verify_key_number(client, 2)
+
+    def test_bloom_deletes(self):
+        client = self.server.get_new_client()
+        # delete
+        assert client.execute_command('BF.ADD filter1 item1') == 1
+        self.verify_bloom_filter_existence(client, 'filter1', 'item1')
+        self.verify_key_number(client, 1)
+        assert client.execute_command('DEL filter1') == 1
+        self.verify_bloom_filter_existence(client, 'filter1', 'item1', should_exist=False)
+        self.verify_key_number(client, 0)
+
+        # flush
+        self.insert_bloom_filter(client, number_of_bf=10)
+        self.verify_key_number(client, 10)
+        assert client.execute_command('FLUSHALL')
+        self.verify_key_number(client, 0)
+
+        # unlink
+        assert client.execute_command('BF.ADD A ITEMA') == 1
+        assert client.execute_command('BF.ADD B ITEMB') == 1
+        self.verify_bloom_filter_existence(client, 'A', 'ITEMA')
+        self.verify_bloom_filter_existence(client, 'B', 'ITEMB')
+        self.verify_bloom_filter_existence(client, 'C', 'ITEMC', should_exist=False)
+        self.verify_key_number(client, 2)
+        assert client.execute_command('UNLINK A B C') == 2
+        assert client.execute_command('BF.MEXISTS A ITEMA ITEMB') == [0, 0]
+        self.verify_bloom_filter_existence(client, 'A', 'ITEMA', should_exist=False)
+        self.verify_bloom_filter_existence(client, 'B', 'ITEMB', should_exist=False)
+        self.verify_key_number(client, 0)
+
+    def test_bloom_expiration(self):
+        client = self.server.get_new_client()
+        # expiration
+        # cmd object idletime
+        self.verify_key_number(client, 0)
+        assert client.execute_command('BF.ADD TEST_IDLE val3') == 1
+        self.verify_bloom_filter_existence(client, 'TEST_IDLE', 'val3')
+        self.verify_key_number(client, 1)
+        time.sleep(1)
+        assert client.execute_command('OBJECT IDLETIME test_idle') == None
+        assert client.execute_command('OBJECT IDLETIME TEST_IDLE') > 0
+        # cmd ttl, expireat
+        assert client.execute_command('BF.ADD TEST_EXP ITEM') == 1
+        assert client.execute_command('TTL TEST_EXP') == -1
+        self.verify_bloom_filter_existence(client, 'TEST_EXP', 'ITEM')
+        self.verify_key_number(client, 2)
+        curr_time = int(time.time())
+        assert client.execute_command(f'EXPIREAT TEST_EXP {curr_time + 5}') == 1
+        wait_for_equal(lambda: client.execute_command('BF.EXISTS TEST_EXP ITEM'), 0)
+        self.verify_key_number(client, 1)
+        # cmd persist
+        assert client.execute_command('BF.ADD TEST_PERSIST ITEM') == 1
+        assert client.execute_command('TTL TEST_PERSIST') == -1
+        self.verify_bloom_filter_existence(client, 'TEST_PERSIST', 'ITEM')
+        self.verify_key_number(client, 2)
+        assert client.execute_command(f'EXPIREAT TEST_PERSIST {curr_time + 100000}') == 1
+        assert client.execute_command('TTL TEST_PERSIST') > 0
+        assert client.execute_command('PERSIST TEST_PERSIST') == 1
+        assert client.execute_command('TTL TEST_PERSIST') == -1

tests/test_bloom_command.py

@@ -0,0 +1,143 @@
+import pytest
+from valkey_bloom_test_case import ValkeyBloomTestCaseBase
+from valkeytests.conftest import resource_port_tracker
+
+class TestBloomCommand(ValkeyBloomTestCaseBase):
+
+    def verify_command_arity(self, command, expected_arity): 
+        command_info = self.client.execute_command('COMMAND', 'INFO', command)
+        actual_arity = command_info.get(command).get('arity')
+        assert actual_arity == expected_arity, f"Arity mismatch for command '{command}'"
+
+    def test_bloom_command_arity(self):
+        self.verify_command_arity('BF.EXISTS', -1)
+        self.verify_command_arity('BF.ADD', -1)
+        self.verify_command_arity('BF.MEXISTS', -1)
+        self.verify_command_arity('BF.MADD', -1)
+        self.verify_command_arity('BF.CARD', -1)
+        self.verify_command_arity('BF.RESERVE', -1)
+        self.verify_command_arity('BF.INFO', -1)
+        self.verify_command_arity('BF.INSERT', -1)
+
+    def test_bloom_command_error(self):
+        # test set up
+        assert self.client.execute_command('BF.ADD key item') == 1
+        assert self.client.execute_command('BF.RESERVE bf 0.01 1000') == b'OK'
+        # non scaling filter
+        assert self.client.execute_command('BF.RESERVE bf_non 0.01 2 NONSCALING') == b'OK'
+        assert self.client.execute_command('BF.ADD bf_non 0') == 1
+        assert self.client.execute_command('BF.ADD bf_non 1') == 1
+
+        basic_error_test_cases = [
+            # not found
+            ('BF.INFO TEST404', 'not found'),
+            # incorrect syntax and argument usage
+            ('BF.ADD bf_non 2', 'non scaling filter is full'),
+            ('bf.info key item', 'invalid information value'),
+            ('bf.insert key CAPACITY 10000 ERROR 0.01 EXPANSION 0.99 NOCREATE NONSCALING ITEMS test1 test2 test3', 'bad expansion'),
+            ('BF.INSERT KEY HELLO WORLD', 'unknown argument received'),
+            ('BF.INSERT KEY error 2 ITEMS test1', '(0 < error rate range < 1)'),
+            ('BF.INSERT TEST_LIMIT ERROR 0.999999999 ITEMS ERROR_RATE', '(0 < error rate range < 1)'),
+            ('BF.INSERT TEST_LIMIT CAPACITY 4394967295 ITEMS CAP', 'bad capacity'),
+            ('BF.INSERT TEST_LIMIT CAPACITY 0 ITEMS CAP0', '(capacity should be larger than 0)'),
+            ('BF.INSERT TEST_LIMIT EXPANSION 11 ITEMS EXPAN', 'bad expansion'),
+            ('BF.INSERT TEST_NOCREATE NOCREATE ITEMS A B', 'not found'),
+            ('BF.RESERVE KEY String 100', 'bad error rate'),
+            ('BF.RESERVE KEY 0.999999999 3000', '(0 < error rate range < 1)'),
+            ('BF.RESERVE KEY 2 100', '(0 < error rate range < 1)'),
+            ('BF.RESERVE KEY 0.01 String', 'bad capacity'),
+            ('BF.RESERVE KEY 0.01 0.01', 'bad capacity'),
+            ('BF.RESERVE KEY 0.01 -1', 'bad capacity'),
+            ('BF.RESERVE KEY 0.01 4394967295', 'bad capacity'),
+            ('BF.RESERVE bf 0.01 1000', 'item exists'),
+            ('BF.RESERVE TEST_CAP 0.50 0', '(capacity should be larger than 0)'),
+
+            # wrong number of arguments
+            ('BF.ADD TEST', 'wrong number of arguments for \'BF.ADD\' command'),
+            ('BF.ADD', 'wrong number of arguments for \'BF.ADD\' command'),
+            ('BF.ADD HELLO TEST WORLD', 'wrong number of arguments for \'BF.ADD\' command'),
+            ('BF.CARD KEY ITEM', 'wrong number of arguments for \'BF.CARD\' command'),
+            ('bf.card', 'wrong number of arguments for \'BF.CARD\' command'),
+            ('BF.EXISTS', 'wrong number of arguments for \'BF.EXISTS\' command'),
+            ('bf.exists item', 'wrong number of arguments for \'BF.EXISTS\' command'),
+            ('bf.exists key item hello', 'wrong number of arguments for \'BF.EXISTS\' command'),
+            ('BF.INFO', 'wrong number of arguments for \'BF.INFO\' command'),
+            ('bf.info key capacity size', 'wrong number of arguments for \'BF.INFO\' command'),
+            ('BF.INSERT', 'wrong number of arguments for \'BF.INSERT\' command'),
+            ('BF.INSERT KEY', 'wrong number of arguments for \'BF.INSERT\' command'),
+            ('BF.INSERT KEY HELLO', 'wrong number of arguments for \'BF.INSERT\' command'),
+            ('BF.INSERT MISS_ITEM EXPANSION 2', 'wrong number of arguments for \'BF.INSERT\' command'),
+            ('BF.MADD', 'wrong number of arguments for \'BF.MADD\' command'),
+            ('BF.MADD KEY', 'wrong number of arguments for \'BF.MADD\' command'),
+            ('BF.MEXISTS', 'wrong number of arguments for \'BF.MEXISTS\' command'),
+            ('BF.MEXISTS INFO', 'wrong number of arguments for \'BF.MEXISTS\' command'),
+            ('BF.RESERVE', 'wrong number of arguments for \'BF.RESERVE\' command'),
+            ('BF.RESERVE KEY', 'wrong number of arguments for \'BF.RESERVE\' command'),
+            ('BF.RESERVE KEY SSS', 'wrong number of arguments for \'BF.RESERVE\' command'),
+            ('BF.RESERVE TT1 0.01 1 NONSCALING test1 test2 test3', 'wrong number of arguments for \'BF.RESERVE\' command'),
+            ('BF.RESERVE TT 0.01 1 NONSCALING EXPANSION 1', 'wrong number of arguments for \'BF.RESERVE\' command'),
+        ]
+
+        for test_case in basic_error_test_cases:
+            cmd = test_case[0]
+            expected_err_reply = test_case[1]
+            self.verify_error_response(self.client, cmd, expected_err_reply)
+
+    def test_bloom_command_behavior(self):
+        basic_behavior_test_case = [
+            ('BF.ADD key item', 1),
+            ('BF.ADD key item', 0),
+            ('BF.ADD key item1', 1),
+            ('BF.EXISTS key item', 1),
+            ('BF.EXISTS key item2', 0),
+            ('BF.MADD key item item2', [0, 1]),
+            ('BF.EXISTS key item', 1),
+            ('BF.EXISTS key item2', 1),
+            ('BF.EXISTS key item3', 0),
+            ('BF.MADD hello world1 world2 world3', [1, 1, 1]),
+            ('BF.MADD hello world1 world2 world3 world4', [0, 0, 0, 1]),
+            ('BF.MEXISTS hello world5', [0]),
+            ('BF.MADD hello world5', [1]),
+            ('BF.MEXISTS hello world5 world6 world7', [1, 0, 0]),
+            ('BF.INSERT TEST ITEMS ITEM', [1]),
+            ('BF.INSERT TEST CAPACITY 1000 ITEMS ITEM', [0]),
+            ('BF.INSERT TEST CAPACITY 200 error 0.50 ITEMS ITEM ITEM1 ITEM2', [0, 1, 1]),
+            ('BF.INSERT TEST CAPACITY 300 ERROR 0.50 EXPANSION 1 ITEMS ITEM FOO', [0, 1]),
+            ('BF.INSERT TEST ERROR 0.50 EXPANSION 3 NOCREATE items BOO', [1]), 
+            ('BF.INSERT TEST ERROR 0.50 EXPANSION 1 NOCREATE NONSCALING items BOO', [0]),
+            ('BF.INSERT TEST_EXPANSION EXPANSION 9 ITEMS ITEM', [1]),
+            ('BF.INSERT TEST_CAPACITY CAPACITY 2000 ITEMS ITEM', [1]),
+            ('BF.INSERT TEST_ITEMS ITEMS 1 2 3 EXPANSION 2', [1, 1, 1, 1, 0]),
+            ('BF.INFO TEST Capacity', 100),
+            ('BF.INFO TEST ITEMS', 5),
+            ('BF.INFO TEST filters', 1),
+            ('bf.info TEST expansion', 2),
+            ('BF.INFO TEST_EXPANSION EXPANSION', 9),
+            ('BF.INFO TEST_CAPACITY CAPACITY', 2000),
+            ('BF.CARD key', 3),
+            ('BF.CARD hello', 5),
+            ('BF.CARD TEST', 5),
+            ('bf.card HELLO', 0),
+            ('BF.RESERVE bf 0.01 1000', b'OK'),
+            ('BF.RESERVE bf_exp 0.01 1000 EXPANSION 2', b'OK'),
+            ('BF.RESERVE bf_non 0.01 1000 NONSCALING', b'OK'),
+            ('bf.info bf_exp expansion', 2),
+            ('BF.INFO bf_non expansion', None),
+        ]
+
+        for test_case in basic_behavior_test_case:
+            cmd = test_case[0]
+            expected_result = test_case[1]
+            self.verify_command_success_reply(self.client, cmd, expected_result)
+
+        # test bf.info
+        assert self.client.execute_command('BF.RESERVE BF_INFO 0.50 2000 NONSCALING') == b'OK'
+        bf_info = self.client.execute_command('BF.INFO BF_INFO')
+        capacity_index = bf_info.index(b'Capacity') + 1
+        filter_index = bf_info.index(b'Number of filters') + 1
+        item_index = bf_info.index(b'Number of items inserted') + 1
+        expansion_index = bf_info.index(b'Expansion rate') + 1
+        assert bf_info[capacity_index] == self.client.execute_command('BF.INFO BF_INFO CAPACITY') == 2000
+        assert bf_info[filter_index] == self.client.execute_command('BF.INFO BF_INFO FILTERS') == 1
+        assert bf_info[item_index] == self.client.execute_command('BF.INFO BF_INFO ITEMS') == 0
+        assert bf_info[expansion_index] == self.client.execute_command('BF.INFO BF_INFO EXPANSION') == None

tests/test_save_and_restore.py

@@ -1,15 +1,9 @@
 import pytest, time
 import os
-from valkey_test_case import ValkeyTestCase
+from valkey_bloom_test_case import ValkeyBloomTestCaseBase
 from valkeytests.conftest import resource_port_tracker
 
-class TestBloomSaveRestore(ValkeyTestCase):
-
-    def get_custom_args(self):
-        self.set_server_version(os.environ['SERVER_VERSION'])
-        return {
-            'loadmodule': os.getenv('MODULE_PATH'),
-        }
+class TestBloomSaveRestore(ValkeyBloomTestCaseBase):
 
     def test_basic_save_and_restore(self):
         client = self.server.get_new_client()