Feat[TESTS]: Add fuzz-tests for subscriptions (#1226)

Signed-off-by: Anton Pryakhin <apryakhin1@bloomberg.net>

Author: waldgange

.github/workflows/build.yaml

@@ -185,6 +185,7 @@ jobs:
           "negotiation_bypass_authentication",
           "negotiation",
           "open_queue",
+          "configure_stream",
           "configure_queue_stream",
           "put",
           "confirm",

src/python/blazingmq/dev/fuzztest/README.md

@@ -9,7 +9,7 @@ handle all messages gracefully, even if message is malformed.
 
 # Launching #
 
-- Build the BlazingMQ broker (the build dir might be `cmake.bld/Linux/src/applications/bmqbrkr`)
+- Build the BlazingMQ broker (the build dir might be `build/blazingmq/src/applications/bmqbrkr`)
 
 - Set up the repo dir for convenience
 
@@ -22,13 +22,13 @@ export BLAZINGMQ_REPO=`pwd`
 ```shell
 python3 -m venv ${BLAZINGMQ_REPO}/venv
 source ${BLAZINGMQ_REPO}/venv/bin/activate
-pip3 install -r "${BLAZINGMQ_REPO}/src/python/blazingmq/dev/requirements.txt"
+pip3 install -r "${BLAZINGMQ_REPO}/src/python/requirements-dev.txt"
 ```
 
 - Launch fuzz testing:
 
 ```shell
 source ${BLAZINGMQ_REPO}/venv/bin/activate
 cd ${BLAZINGMQ_REPO}/src/python
-python3 -m blazingmq.dev.fuzztest --broker-dir "${BLAZINGMQ_REPO}/cmake.bld/Linux/src/applications/bmqbrkr"
+python3 -m blazingmq.dev.fuzztest --broker-dir "${BLAZINGMQ_REPO}/build/blazingmq/src/applications/bmqbrkr"
 ```

src/python/blazingmq/dev/fuzztest/__init__.py

@@ -110,6 +110,15 @@ def schema_to_boofuzz(schema: broker.SchemaDescription) -> BoofuzzSequence:
 
         if isinstance(value, dict):
             res += schema_to_boofuzz(value)
+        elif isinstance(value, list):
+            res.append(boofuzz.Static(default_value="["))
+            for item in value:
+                res += schema_to_boofuzz(item)
+                res.append(boofuzz.Delim(default_value=",", fuzzable=fuzz_delimiters))
+            # last boofuzz.Delim "," should be removed if presented
+            if len(value) > 0:
+                res = res[:-1]
+            res.append(boofuzz.Static(default_value="]"))
         elif isinstance(value, str):
             res.append(boofuzz.String(default_value=f'"{value}"'))
         elif isinstance(value, bool):
@@ -397,6 +406,11 @@ def fuzz(host: str, port: int, request: Optional[str] = None) -> None:
         "OpenQueue", children=(make_control_message(broker.OPEN_QUEUE_SCHEMA))
     )
 
+    configure_stream = boofuzz.Request(
+        "ConfigureStream",
+        children=(make_control_message(broker.CONFIGURE_STREAM_SCHEMA)),
+    )
+
     configure_queue_stream = boofuzz.Request(
         "ConfigureQueueStream",
         children=(make_control_message(broker.CONFIGURE_QUEUE_STREAM_SCHEMA)),
@@ -418,6 +432,7 @@ def fuzz(host: str, port: int, request: Optional[str] = None) -> None:
         (authentication, "authentication"),
         (negotiation, "negotiation"),
         (open_queue, "open_queue"),
+        (configure_stream, "configure_stream"),
         (configure_queue_stream, "configure_queue_stream"),
         (put, "put"),
         (confirm, "confirm"),

src/python/blazingmq/schemas/broker.py

@@ -103,6 +103,33 @@ class PutHeaderFlags(IntFlag):
     }
 }
 
+CONFIGURE_STREAM_SCHEMA: SchemaDescription = {
+    "rId": 0,
+    "configureStream": {
+        "qId": 0,
+        "streamParameters": {
+            "appId": "__default",
+            "subscriptions": [
+                {
+                    "sId": 0,
+                    "expression": {
+                        "version": "E_VERSION_1",
+                        "text": "test_expression",
+                    },
+                    "consumers": [
+                        {
+                            "maxUnconfirmedMessages": 0,
+                            "maxUnconfirmedBytes": 0,
+                            "consumerPriority": -2147483648,
+                            "consumerPriorityCount": 0,
+                        },
+                    ],
+                },
+            ],
+        },
+    },
+}
+
 CONFIGURE_QUEUE_STREAM_SCHEMA: SchemaDescription = {
     "rId": 0,
     "configureQueueStream": {