Adding a bloom command keyspace test

Signed-off-by: zackcam <[email protected]>

tests/test_bloom_keyspace.py

@@ -0,0 +1,64 @@
+import logging, time
+from valkey_bloom_test_case import ValkeyBloomTestCaseBase
+from valkeytests.conftest import resource_port_tracker
+
+class TestKeyEventNotifications(ValkeyBloomTestCaseBase):
+    RESERVE_KEYSPACE_MESSAGE = {'type': 'pmessage', 'pattern': b'__key*__:*', 'channel': b'__keyspace@0__:intermediate_val', 'data': b'bloom.reserve'}
+    RESERVE_KEYEVENT_MESSAGE = {'type': 'pmessage', 'pattern': b'__key*__:*', 'channel': b'__keyevent@0__:bloom.reserve', 'data': b'intermediate_val'}
+    ADD_KEYSPACE_MESSAGE = {'type': 'pmessage', 'pattern': b'__key*__:*', 'channel': b'__keyspace@0__:intermediate_val', 'data': b'bloom.add'} 
+    ADD_KEYEVENT_MESSAGE = {'type': 'pmessage', 'pattern': b'__key*__:*', 'channel': b'__keyevent@0__:bloom.add', 'data': b'intermediate_val'}
+
+    def create_expected_message_list(self, reserve_expected, add_expected, key_name):
+        expected_messages = []
+        self.RESERVE_KEYSPACE_MESSAGE['channel'] = f"__keyspace@0__:{key_name}".encode('utf-8')
+        self.RESERVE_KEYEVENT_MESSAGE['data'] = f"{key_name}".encode('utf-8')
+        self.ADD_KEYSPACE_MESSAGE['channel'] = f"__keyspace@0__:{key_name}".encode('utf-8')
+        self.ADD_KEYEVENT_MESSAGE['data'] = f"{key_name}".encode('utf-8')
+        if reserve_expected:
+            expected_messages.append(self.RESERVE_KEYEVENT_MESSAGE)
+            expected_messages.append(self.RESERVE_KEYSPACE_MESSAGE)
+        if add_expected:
+            expected_messages.append(self.ADD_KEYSPACE_MESSAGE)
+            expected_messages.append(self.ADD_KEYEVENT_MESSAGE)
+        return expected_messages
+
+    def check_response(self, result_messages, expected_messages):
+        assert len(expected_messages) == len(result_messages), f"Number of messages received differs, Expected messages: {expected_messages}, Messages received: {result_messages}"
+        for message in expected_messages:
+            assert message in result_messages, f"{message} was not found in messages received"
+
+    def get_subscribe_client_messages(self, client, cmd):
+        client.execute_command(cmd)
+        count = 0
+        messages = []
+        timeout = time.time() + 5
+        while time.time() < timeout:
+            message = self.keyspace_client_subscribe.get_message()
+            if message:
+                # Only for the first time we get messages we should skip the first message gotten
+                if count > 0 or "BF.ADD" not in cmd:
+                    messages.append(message)    
+                count = count + 1
+        return messages
+
+    def test_keyspace_bloom_commands(self):
+        self.create_subscribe_clients()
+        bloom_commands = [
+            ('BF.ADD add_test key', True, True),
+            ('BF.MADD madd_test key1 key2', True, True),
+            ('BF.EXISTS exists_test key', False, False),
+            ('BF.INSERT insert_test ITEMS key1 key2', True, True),
+            ('BF.RESERVE reserve_test 0.01 1000', True, False)
+        ]
+
+        for command, reserve_expected, add_expected in bloom_commands:
+            expected_messages = self.create_expected_message_list(reserve_expected, add_expected, command.split()[1]) if reserve_expected else []
+            result_messages = self.get_subscribe_client_messages(self.keyspace_client, command)
+            self.check_response(result_messages, expected_messages)
+
+    def create_subscribe_clients(self):
+        self.keyspace_client = self.server.get_new_client()
+        self.keyspace_client_subscribe = self.keyspace_client.pubsub()
+        self.keyspace_client_subscribe.psubscribe('__key*__:*')
+        self.keyspace_client.execute_command('CONFIG' ,'SET','notify-keyspace-events', 'KEA')
+